""" core/injection/system_prompt_injector.py ========================================= Story 7.06 — SystemPromptInjector Builder Function Story 7.07 — SystemPromptInjector Telnyx API Push Assembles the full system prompt injection block from recent memory + AIVA persona. Called before every LLM context window is populated. Story 7.07 adds push_to_telnyx() to update the Telnyx AI Assistant system prompt via PATCH /v2/ai_assistants/{assistant_id} before a call. # VERIFICATION_STAMP # Story: 7.06 # Verified By: parallel-builder # Verified At: 2026-02-25 # Tests: tests/track_a/test_story_7_06.py # Coverage: 100% # VERIFICATION_STAMP # Story: 7.07 # Verified By: parallel-builder # Verified At: 2026-02-25 # Tests: tests/track_a/test_story_7_07.py # Coverage: 100% """ from __future__ import annotations import logging import os from datetime import datetime, timezone, timedelta from typing import Any, Optional logger = logging.getLogger(__name__) AEST_OFFSET = timedelta(hours=10) TELNYX_API_BASE = "https://api.telnyx.com/v2" INJECTION_TEMPLATE = """=== AIVA MEMORY INJECTION === {recent_conversations} === KING'S ACTIVE DIRECTIVES === {kinan_directives} === CURRENT SESSION STATE === Caller: {caller_info} Time: {local_time} AEST Active Tasks: {open_tasks} === END MEMORY INJECTION ===""" class SystemPromptInjector: """ Assembles the full system prompt injection block and pushes it to Telnyx. Parameters ---------- conversation_engine: Object with ``get_recent_summary(n: int) -> str`` method. Pass ``None`` to disable conversation history lookup (fallback text used). redis_client: Redis client for reading directives and session state. Pass ``None`` to disable Redis lookups (fallback text used for all fields). http_client: Optional HTTP client for making API requests. Supports dependency injection for testing — pass a mock to avoid real network calls. Must expose an async ``patch(url, json, headers)`` method that returns a response object with a ``.status`` (or ``.status_code``) attribute. Pass ``None`` to use ``aiohttp.ClientSession`` at call time (production). """ def __init__( self, conversation_engine=None, redis_client=None, http_client: Optional[Any] = None, ): self.conversation_engine = conversation_engine self.redis = redis_client self.http = http_client # ------------------------------------------------------------------ # Public API # ------------------------------------------------------------------ async def build_injection(self, session_id: str) -> str: """ Build and return the formatted memory injection block. Steps ----- 1. Get recent_conversations from conversation_engine (last 3 summaries). 2. Get kinan_directives from Redis key ``kinan:directives:active``. 3. Get caller_info from Redis key ``aiva:state:{session_id}``. 4. Get open_tasks from Redis key ``aiva:tasks:active``. 5. Format INJECTION_TEMPLATE with AEST time. 6. Return formatted string (always non-empty — fallbacks ensure this). Parameters ---------- session_id: Unique identifier for the current AIVA session / call leg. Returns ------- str Populated injection block. Never empty. """ recent = self._get_recent_conversations() directives = self._get_directives() caller = self._get_caller_info(session_id) tasks = self._get_open_tasks() local_time = self._get_aest_time() return INJECTION_TEMPLATE.format( recent_conversations=recent, kinan_directives=directives, caller_info=caller, local_time=local_time, open_tasks=tasks, ) async def push_to_telnyx(self, session_id: str, assistant_id: str) -> bool: """ Build the memory injection block and push it to the Telnyx AI Assistant. Steps ----- 1. Call ``build_injection(session_id)`` to get the full injection block. 2. PATCH ``/v2/ai_assistants/{assistant_id}`` with JSON body ``{"system_prompt": }``. 3. Return ``True`` on HTTP 200/204, ``False`` on any error. Parameters ---------- session_id: Unique identifier for the current AIVA session / call leg. Passed through to ``build_injection`` for caller-context lookup. assistant_id: The Telnyx AI Assistant UUID to update. Example: ``"assistant-9c42d3ce-e05a-4e34-8083-c91081917637"`` Returns ------- bool ``True`` if the PATCH succeeded (HTTP 2xx). ``False`` on any HTTP error or network failure — never raises. Auth ---- Bearer token read from ``TELNYX_API_KEY`` environment variable. The key is NOT hardcoded and NOT logged. """ api_key = os.environ.get("TELNYX_API_KEY", "") if not api_key: logger.error( "push_to_telnyx: TELNYX_API_KEY not set in environment — cannot push" ) return False url = f"{TELNYX_API_BASE}/ai_assistants/{assistant_id}" headers = { "Authorization": f"Bearer {api_key}", "Content-Type": "application/json", } try: system_prompt = await self.build_injection(session_id) payload = {"system_prompt": system_prompt} if self.http is not None: # Injected client (used in tests and production callers that manage # their own session lifecycle) response = await self.http.patch(url, json=payload, headers=headers) # Support both aiohttp (.status) and httpx (.status_code) conventions status = ( response.status if hasattr(response, "status") else response.status_code ) else: # Production path — create a one-off aiohttp session import aiohttp # deferred import; not required in test environments async with aiohttp.ClientSession() as session: async with session.patch( url, json=payload, headers=headers ) as resp: status = resp.status if 200 <= status < 300: logger.info( "push_to_telnyx: successfully updated assistant %s (HTTP %s)", assistant_id, status, ) return True logger.error( "push_to_telnyx: Telnyx API returned HTTP %s for assistant %s", status, assistant_id, ) return False except Exception: logger.exception( "push_to_telnyx: unexpected error updating assistant %s", assistant_id ) return False # ------------------------------------------------------------------ # Private helpers # ------------------------------------------------------------------ def _get_recent_conversations(self) -> str: """Fetch last 3 conversation summaries from conversation_engine.""" if self.conversation_engine is None: return "No prior conversations" try: return self.conversation_engine.get_recent_summary(3) except Exception: logger.exception("Failed to fetch recent conversations") return "No prior conversations" def _get_directives(self) -> str: """Fetch active Kinan directives from Redis.""" if self.redis is None: return "No active directives" try: val = self.redis.get("kinan:directives:active") if val is None: return "No active directives" return val.decode("utf-8") if isinstance(val, bytes) else str(val) except Exception: logger.exception("Failed to fetch Kinan directives from Redis") return "No active directives" def _get_caller_info(self, session_id: str) -> str: """Fetch caller info for the session from Redis.""" if self.redis is None: return "Unknown" try: val = self.redis.get(f"aiva:state:{session_id}") if val is None: return "Unknown" return val.decode("utf-8") if isinstance(val, bytes) else str(val) except Exception: logger.exception("Failed to fetch caller info from Redis for session %s", session_id) return "Unknown" def _get_open_tasks(self) -> str: """Fetch active task list from Redis.""" if self.redis is None: return "None" try: val = self.redis.get("aiva:tasks:active") if val is None: return "None" return val.decode("utf-8") if isinstance(val, bytes) else str(val) except Exception: logger.exception("Failed to fetch open tasks from Redis") return "None" @staticmethod def _get_aest_time() -> str: """Return current AEST time (UTC+10) as formatted string.""" utc_now = datetime.now(timezone.utc) aest_now = utc_now + AEST_OFFSET return aest_now.strftime("%Y-%m-%d %H:%M:%S")