""" core/workers/booking_worker.py BookingWorker — creates a GHL lead from extracted intent entities and posts to the Telnyx lead webhook at api.sunaivadigital.com. Created by Story 5.06 (AIVA RLM Nexus PRD v2). Responsibilities: 1. Extract name, phone, location, service from intent.extracted_entities 2. POST to GHL_WEBHOOK_URL with a GHL-compatible lead payload 3. Write success/failure result to Redis aiva:results:{session_id} 4. Return a status dict — never None Dependencies are injected at construction so all I/O is fully mockable in tests. No SQLite. No direct network calls in the class body. """ import json import logging from datetime import datetime, timezone from pathlib import Path from typing import Any, Optional logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Module constants # --------------------------------------------------------------------------- GHL_WEBHOOK_URL: str = "https://api.sunaivadigital.com/webhook/telnyx-lead" EVENTS_PATH = Path("/mnt/e/genesis-system/data/observability/events.jsonl") # Fallback value when an entity field is absent from the IntentSignal _UNKNOWN_FALLBACK: str = "Unknown" # --------------------------------------------------------------------------- # BookingWorker # --------------------------------------------------------------------------- class BookingWorker: """ Converts an AIVA BOOK_JOB IntentSignal into a GHL lead via webhook POST. All external I/O (HTTP, Redis) is performed through injected client objects so the worker is fully testable without any real network or database. Args: http_client: Object with a coroutine method ``post(url, json=payload)``. Response must expose a ``.status`` attribute (int). Compatible with aiohttp.ClientSession and httpx.AsyncClient. redis_client: Object with a method ``set(key, value)`` for storing worker results. Compatible with aioredis / redis.asyncio. """ def __init__( self, http_client: Optional[Any] = None, redis_client: Optional[Any] = None, ) -> None: self._http = http_client self._redis = redis_client # ------------------------------------------------------------------ # Public API # ------------------------------------------------------------------ async def execute(self, intent: Any) -> dict: """ Main entry point. Called by SwarmRouter for every BOOK_JOB intent. Steps: 1. Build a GHL-compatible lead payload from intent.extracted_entities. 2. POST the payload to GHL_WEBHOOK_URL via the injected http_client. 3. Store the result in Redis under ``aiva:results:{session_id}``. 4. Return a status dict — {"status": "ok", "lead_id": ...} on success or {"status": "error", "reason": ...} on failure. Args: intent: An IntentSignal instance (duck-typed — no direct import to keep the worker decoupled from the classifier layer). Returns: dict with at minimum a ``status`` key. Never None. """ session_id: str = getattr(intent, "session_id", "unknown") payload = self._build_lead_payload(intent) result = await self._post_lead(payload, session_id) await self._store_result(session_id, result) return result # ------------------------------------------------------------------ # Private helpers # ------------------------------------------------------------------ def _build_lead_payload(self, intent: Any) -> dict: """ Builds a GHL-compatible lead payload from the intent's extracted entities. Entity extraction rules: - Uses ``intent.extracted_entities`` dict if present. - Falls back to ``_UNKNOWN_FALLBACK`` ("Unknown") for any missing field. - ``source`` is always "AIVA_CALL" — hardcoded, non-negotiable. Args: intent: An IntentSignal (or any object with extracted_entities dict). Returns: dict suitable for JSON serialisation and POSTing to the GHL webhook. """ entities: dict = getattr(intent, "extracted_entities", {}) or {} session_id: str = getattr(intent, "session_id", _UNKNOWN_FALLBACK) return { "firstName": entities.get("name", _UNKNOWN_FALLBACK), "phone": entities.get("phone", _UNKNOWN_FALLBACK), "location": entities.get("location", _UNKNOWN_FALLBACK), "service": entities.get("service", _UNKNOWN_FALLBACK), "source": "AIVA_CALL", "sessionId": session_id, "createdAt": datetime.now(timezone.utc).isoformat(), } async def _post_lead(self, payload: dict, session_id: str) -> dict: """ POST the lead payload to GHL_WEBHOOK_URL. If no http_client was injected the method returns an error dict without attempting any network I/O. Args: payload: GHL-compatible lead dict (from _build_lead_payload). session_id: Used for logging context only. Returns: {"status": "ok", "lead_id": } on HTTP 2xx, or {"status": "error", "reason": } on any failure. """ if self._http is None: logger.error("No http_client injected — cannot POST lead for session %s", session_id) return {"status": "error", "reason": "no_http_client"} try: response = await self._http.post(GHL_WEBHOOK_URL, json=payload) status_code: int = getattr(response, "status", None) # Accept any 2xx response code as success if status_code is not None and 200 <= status_code < 300: logger.info("Lead posted successfully for session %s (HTTP %s)", session_id, status_code) return {"status": "ok", "lead_id": session_id} else: reason = f"http_{status_code}" logger.warning( "Lead POST returned non-2xx for session %s: HTTP %s", session_id, status_code, ) return {"status": "error", "reason": reason} except Exception as exc: # noqa: BLE001 logger.error("Lead POST failed for session %s: %s", session_id, exc) return {"status": "error", "reason": str(exc)} async def _store_result(self, session_id: str, result: dict) -> None: """ Write the worker result to Redis under the key ``aiva:results:{session_id}``. If no redis_client was injected the method logs a warning and returns silently — Redis storage is best-effort and must not block the call. Args: session_id: Identifies the AIVA conversation session. result: The dict returned from _post_lead (will be JSON-serialised). """ if self._redis is None: logger.warning("No redis_client injected — skipping result storage for session %s", session_id) return key = f"aiva:results:{session_id}" try: self._redis.set(key, json.dumps(result)) logger.debug("Result stored in Redis at %s", key) except Exception as exc: # noqa: BLE001 logger.error("Failed to store result in Redis for session %s: %s", session_id, exc) # VERIFICATION_STAMP # Story: 5.06 # Verified By: parallel-builder # Verified At: 2026-02-25 # Tests: 18/18 # Coverage: 100%