Merge branch 'main' into feature/before-guardrail

src/va_agent/agent.py

@@ -1,38 +1,63 @@
 """ADK agent with vector search RAG tool."""
 
+from functools import partial
+
 from google import genai
 from google.adk.agents.llm_agent import Agent
 from google.adk.runners import Runner
 from google.adk.tools.mcp_tool import McpToolset
 from google.adk.tools.mcp_tool.mcp_session_manager import StreamableHTTPConnectionParams
 from google.cloud.firestore_v1.async_client import AsyncClient
+from google.genai.types import Content, Part
 
 from va_agent.auth import auth_headers_provider
 from va_agent.config import settings
-from va_agent.session import FirestoreSessionService
+from va_agent.dynamic_instruction import provide_dynamic_instruction
 from va_agent.governance import GovernancePlugin
+from va_agent.notifications import FirestoreNotificationBackend
+from va_agent.session import FirestoreSessionService
 
+# MCP Toolset for RAG knowledge search
 toolset = McpToolset(
     connection_params=StreamableHTTPConnectionParams(url=settings.mcp_remote_url),
     header_provider=auth_headers_provider,
 )
 
+
+# Shared Firestore client for session service and notifications
+firestore_db = AsyncClient(database=settings.firestore_db)
+
+# Session service with compaction
+session_service = FirestoreSessionService(
+    db=firestore_db,
+    compaction_token_threshold=10_000,
+    genai_client=genai.Client(),
+)
+
+# Notification service
+notification_service = FirestoreNotificationBackend(
+    db=firestore_db,
+    collection_path=settings.notifications_collection_path,
+    max_to_notify=settings.notifications_max_to_notify,
+    window_hours=settings.notifications_window_hours,
+)
+
+# Agent with static and dynamic instructions
 governance = GovernancePlugin()
 agent = Agent(
     model=settings.agent_model,
     name=settings.agent_name,
-    instruction=settings.agent_instructions,
+    instruction=partial(provide_dynamic_instruction, notification_service),
+    static_instruction=Content(
+        role="user",
+        parts=[Part(text=settings.agent_instructions)],
+    ),
     tools=[toolset],
     before_model_callback=governance.before_model_callback,
     after_model_callback=governance.after_model_callback,
 )
 
-session_service = FirestoreSessionService(
-    db=AsyncClient(database=settings.firestore_db),
-    compaction_token_threshold=10_000,
-    genai_client=genai.Client(),
-)
-
+# Runner
 runner = Runner(
     app_name="va_agent",
     agent=agent,

src/va_agent/config.py

@@ -26,6 +26,13 @@ class AgentSettings(BaseSettings):
     # Firestore configuration
     firestore_db: str
 
+    # Notifications configuration
+    notifications_collection_path: str = (
+        "artifacts/bnt-orquestador-cognitivo-dev/notifications"
+    )
+    notifications_max_to_notify: int = 5
+    notifications_window_hours: float = 48
+
     # MCP configuration
     mcp_audience: str
     mcp_remote_url: str
@@ -33,7 +40,7 @@ class AgentSettings(BaseSettings):
     model_config = SettingsConfigDict(
         yaml_file=CONFIG_FILE_PATH,
         extra="ignore",  # Ignore extra fields from config.yaml
-        env_file=".env"
+        env_file=".env",
     )
 
     @classmethod

src/va_agent/dynamic_instruction.py

@@ -0,0 +1,134 @@
+"""Dynamic instruction provider for VAia agent."""
+
+from __future__ import annotations
+
+import logging
+import time
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from google.adk.agents.readonly_context import ReadonlyContext
+
+    from va_agent.notifications import NotificationBackend
+
+logger = logging.getLogger(__name__)
+
+
+_SECONDS_PER_MINUTE = 60
+_SECONDS_PER_HOUR = 3600
+_MINUTES_PER_HOUR = 60
+_HOURS_PER_DAY = 24
+
+
+def _format_time_ago(now: float, ts: float) -> str:
+    """Return a human-readable Spanish label like 'hace 3 horas'."""
+    diff = max(now - ts, 0)
+    minutes = int(diff // _SECONDS_PER_MINUTE)
+    hours = int(diff // _SECONDS_PER_HOUR)
+
+    if minutes < 1:
+        return "justo ahora"
+    if minutes < _MINUTES_PER_HOUR:
+        return f"hace {minutes} min"
+    if hours < _HOURS_PER_DAY:
+        return f"hace {hours}h"
+    days = hours // _HOURS_PER_DAY
+    return f"hace {days}d"
+
+
+async def provide_dynamic_instruction(
+    notification_service: NotificationBackend,
+    ctx: ReadonlyContext | None = None,
+) -> str:
+    """Provide dynamic instructions based on recent notifications.
+
+    This function is called by the ADK agent on each message. It:
+    1. Queries Firestore for recent notifications
+    2. Marks them as notified
+    3. Returns a dynamic instruction for the agent to mention them
+
+    Args:
+        notification_service: Service for fetching/marking notifications
+        ctx: Agent context containing session information
+
+    Returns:
+        Dynamic instruction string (empty if no notifications or not first message)
+
+    """
+    # Only check notifications on the first message
+    if not ctx:
+        logger.debug("No context available for dynamic instruction")
+        return ""
+
+    session = ctx.session
+    if not session:
+        logger.debug("No session available for dynamic instruction")
+        return ""
+
+    # Extract phone number from user_id (they are the same in this implementation)
+    phone_number = session.user_id
+
+    logger.info(
+        "Checking recent notifications for user %s",
+        phone_number,
+    )
+
+    try:
+        # Fetch recent notifications
+        recent_notifications = await notification_service.get_recent_notifications(
+            phone_number
+        )
+
+        if not recent_notifications:
+            logger.info("No recent notifications for user %s", phone_number)
+            return ""
+
+        # Build dynamic instruction with notification details
+        notification_ids = [
+            nid
+            for n in recent_notifications
+            if (nid := n.get("id_notificacion")) is not None
+        ]
+        count = len(recent_notifications)
+
+        # Format notification details for the agent (most recent first)
+        now = time.time()
+        notification_details = []
+        for i, notif in enumerate(recent_notifications, 1):
+            evento = notif.get("nombre_evento_dialogflow", "notificacion")
+            texto = notif.get("texto", "Sin texto")
+            ts = notif.get("timestamp_creacion", notif.get("timestampCreacion", 0))
+            ago = _format_time_ago(now, ts)
+            notification_details.append(
+                f"  {i}. [{ago}] Evento: {evento} | Texto: {texto}"
+            )
+
+        details_text = "\n".join(notification_details)
+
+        header = (
+            f"Estas son {count} notificación(es) reciente(s)"
+            " de las cuales el usuario podría preguntar más:"
+        )
+        instruction = f"""
+{header}
+
+{details_text}
+"""
+
+        # Mark notifications as notified in Firestore
+        await notification_service.mark_as_notified(phone_number, notification_ids)
+
+        logger.info(
+            "Returning dynamic instruction with %d notification(s) for user %s",
+            count,
+            phone_number,
+        )
+
+    except Exception:
+        logger.exception(
+            "Error building dynamic instruction for user %s",
+            phone_number,
+        )
+        return ""
+    else:
+        return instruction

src/va_agent/governance.py

@@ -106,6 +106,7 @@ Devuelve un JSON con la siguiente estructura:
             rf"|🖕{tone_pattern}"  # middle finger with all skin tone variations
         )
 
+
     def _remove_emojis(self, text: str) -> tuple[str, list[str]]:
         removed = self._combined_pattern.findall(text)
         text = self._combined_pattern.sub("", text)

src/va_agent/notifications.py

@@ -0,0 +1,232 @@
+"""Notification management for VAia agent."""
+
+from __future__ import annotations
+
+import logging
+import time
+from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable
+
+if TYPE_CHECKING:
+    from google.cloud.firestore_v1.async_client import AsyncClient
+
+logger = logging.getLogger(__name__)
+
+
+@runtime_checkable
+class NotificationBackend(Protocol):
+    """Backend-agnostic interface for notification storage."""
+
+    async def get_recent_notifications(self, phone_number: str) -> list[dict[str, Any]]:
+        """Return recent notifications for *phone_number*."""
+        ...
+
+    async def mark_as_notified(
+        self, phone_number: str, notification_ids: list[str]
+    ) -> bool:
+        """Mark the given notification IDs as notified. Return success."""
+        ...
+
+
+class FirestoreNotificationBackend:
+    """Firestore-backed notification backend (read-only).
+
+    Reads notifications from a Firestore document keyed by phone number.
+    Filters by a configurable time window instead of tracking read/unread
+    state — the agent is awareness-only; delivery happens in the app.
+    """
+
+    def __init__(
+        self,
+        *,
+        db: AsyncClient,
+        collection_path: str,
+        max_to_notify: int = 5,
+        window_hours: float = 48,
+    ) -> None:
+        """Initialize with Firestore client and collection path."""
+        self._db = db
+        self._collection_path = collection_path
+        self._max_to_notify = max_to_notify
+        self._window_hours = window_hours
+
+    async def get_recent_notifications(self, phone_number: str) -> list[dict[str, Any]]:
+        """Get recent notifications for a user.
+
+        Retrieves notifications created within the configured time window,
+        ordered by timestamp (most recent first), limited to max_to_notify.
+
+        Args:
+            phone_number: User's phone number (used as document ID)
+
+        Returns:
+            List of notification dictionaries with structure:
+            {
+                "id_notificacion": str,
+                "texto": str,
+                "status": str,
+                "timestamp_creacion": timestamp,
+                "parametros": {...}
+            }
+
+        """
+        try:
+            doc_ref = self._db.collection(self._collection_path).document(phone_number)
+            doc = await doc_ref.get()
+
+            if not doc.exists:
+                logger.info(
+                    "No notification document found for phone: %s", phone_number
+                )
+                return []
+
+            data = doc.to_dict() or {}
+            all_notifications = data.get("notificaciones", [])
+
+            if not all_notifications:
+                logger.info("No notifications in array for phone: %s", phone_number)
+                return []
+
+            cutoff = time.time() - (self._window_hours * 3600)
+
+            def _ts(n: dict[str, Any]) -> Any:
+                return n.get(
+                    "timestamp_creacion",
+                    n.get("timestampCreacion", 0),
+                )
+
+            recent = [n for n in all_notifications if _ts(n) >= cutoff]
+
+            if not recent:
+                logger.info(
+                    "No notifications within the last %.0fh for phone: %s",
+                    self._window_hours,
+                    phone_number,
+                )
+                return []
+
+            recent.sort(key=_ts, reverse=True)
+
+            result = recent[: self._max_to_notify]
+
+            logger.info(
+                "Found %d recent notifications for phone: %s (returning top %d)",
+                len(recent),
+                phone_number,
+                len(result),
+            )
+
+        except Exception:
+            logger.exception(
+                "Failed to fetch notifications for phone: %s", phone_number
+            )
+            return []
+        else:
+            return result
+
+    async def mark_as_notified(
+        self,
+        phone_number: str,  # noqa: ARG002
+        notification_ids: list[str],  # noqa: ARG002
+    ) -> bool:
+        """No-op — the agent is not the delivery mechanism."""
+        return True
+
+
+class RedisNotificationBackend:
+    """Redis-backed notification backend (read-only)."""
+
+    def __init__(
+        self,
+        *,
+        host: str = "127.0.0.1",
+        port: int = 6379,
+        max_to_notify: int = 5,
+        window_hours: float = 48,
+    ) -> None:
+        """Initialize with Redis connection parameters."""
+        import redis.asyncio as aioredis  # noqa: PLC0415
+
+        self._client = aioredis.Redis(
+            host=host,
+            port=port,
+            decode_responses=True,
+            socket_connect_timeout=5,
+        )
+        self._max_to_notify = max_to_notify
+        self._window_hours = window_hours
+
+    async def get_recent_notifications(self, phone_number: str) -> list[dict[str, Any]]:
+        """Get recent notifications for a user from Redis.
+
+        Reads from the ``notification:{phone}`` key, parses the JSON
+        payload, and returns notifications created within the configured
+        time window, sorted by creation timestamp (most recent first),
+        limited to *max_to_notify*.
+        """
+        import json  # noqa: PLC0415
+
+        try:
+            raw = await self._client.get(f"notification:{phone_number}")
+
+            if not raw:
+                logger.info(
+                    "No notification data in Redis for phone: %s",
+                    phone_number,
+                )
+                return []
+
+            data = json.loads(raw)
+            all_notifications: list[dict[str, Any]] = data.get("notificaciones", [])
+
+            if not all_notifications:
+                logger.info(
+                    "No notifications in array for phone: %s",
+                    phone_number,
+                )
+                return []
+
+            cutoff = time.time() - (self._window_hours * 3600)
+
+            def _ts(n: dict[str, Any]) -> Any:
+                return n.get(
+                    "timestamp_creacion",
+                    n.get("timestampCreacion", 0),
+                )
+
+            recent = [n for n in all_notifications if _ts(n) >= cutoff]
+
+            if not recent:
+                logger.info(
+                    "No notifications within the last %.0fh for phone: %s",
+                    self._window_hours,
+                    phone_number,
+                )
+                return []
+
+            recent.sort(key=_ts, reverse=True)
+
+            result = recent[: self._max_to_notify]
+
+            logger.info(
+                "Found %d recent notifications for phone: %s (returning top %d)",
+                len(recent),
+                phone_number,
+                len(result),
+            )
+
+        except Exception:
+            logger.exception(
+                "Failed to fetch notifications from Redis for phone: %s",
+                phone_number,
+            )
+            return []
+        else:
+            return result
+
+    async def mark_as_notified(
+        self,
+        phone_number: str,  # noqa: ARG002
+        notification_ids: list[str],  # noqa: ARG002
+    ) -> bool:
+        """No-op — the agent is not the delivery mechanism."""
+        return True

src/va_agent/server.py

@@ -22,20 +22,11 @@ app = FastAPI(title="Vaia Agent")
 # ---------------------------------------------------------------------------
 
 
-class NotificationPayload(BaseModel):
-    """Notification context sent alongside a user query."""
-
-    text: str | None = None
-    parameters: dict[str, Any] = Field(default_factory=dict)
-
-
 class QueryRequest(BaseModel):
     """Incoming query request from the integration layer."""
 
     phone_number: str
     text: str
-    type: str = "conversation"
-    notification: NotificationPayload | None = None
     language_code: str = "es"
 
 
@@ -56,26 +47,6 @@ class ErrorResponse(BaseModel):
     status: int
 
 
-# ---------------------------------------------------------------------------
-# Helpers
-# ---------------------------------------------------------------------------
-
-
-def _build_user_message(request: QueryRequest) -> str:
-    """Compose the text sent to the agent, including notification context."""
-    if request.type == "notification" and request.notification:
-        parts = [request.text]
-        if request.notification.text:
-            parts.append(f"\n[Notificación recibida]: {request.notification.text}")
-        if request.notification.parameters:
-            formatted = ", ".join(
-                f"{k}: {v}" for k, v in request.notification.parameters.items()
-            )
-            parts.append(f"[Parámetros de notificación]: {formatted}")
-        return "\n".join(parts)
-    return request.text
-
-
 # ---------------------------------------------------------------------------
 # Endpoints
 # ---------------------------------------------------------------------------
@@ -92,13 +63,12 @@ def _build_user_message(request: QueryRequest) -> str:
 )
 async def query(request: QueryRequest) -> QueryResponse:
     """Process a user message and return a generated response."""
-    user_message = _build_user_message(request)
     session_id = request.phone_number
     user_id = request.phone_number
 
     new_message = Content(
         role="user",
-        parts=[Part(text=user_message)],
+        parts=[Part(text=request.text)],
     )
 
     try: