'i&dZddlmZddlZddlZddlmZmZmZddlm Z m Z eje Z edZdZd ZGd d Zy) u 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% ) annotationsN)datetimetimezone timedelta)AnyOptional )hourszhttps://api.telnyx.com/v2z=== 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 ===cdeZdZdZ d d dZd dZddZddZddZd dZ dd Z e dd Z y)SystemPromptInjectoruZ 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). Nc.||_||_||_y)N)conversation_engineredishttp)selfr redis_client http_clients >/mnt/e/genesis-system/core/injection/system_prompt_injector.py__init__zSystemPromptInjector.__init__Hs $7 !  cK|j}|j}|j|}|j}|j }t j |||||Sw)u 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_conversationskinan_directives caller_info local_time open_tasks)_get_recent_conversations_get_directives_get_caller_info_get_open_tasks_get_aest_timeINJECTION_TEMPLATEformat)r session_idrecent directivescallertasksrs rbuild_injectionz$SystemPromptInjector.build_injectionVsw.//1))+ &&z2$$&((* !((!''! )  sA,A.cKtjjdd}|stj dyt d|}d|dd} |j |d {}d |i}|jK|jj||| d {}t|d r |jn |j} nfd d l } | j4d {} | j||| 4d {} | j} d d d d {d d d d {d cxkrdkrnntjd|| ytj d| |y7777u7[#1d {7swYkxYw7b#1d {7swYrxYw#t$rtj!d|YywxYww)u0 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. TELNYX_API_KEYuEpush_to_telnyx: TELNYX_API_KEY not set in environment — cannot pushFz/ai_assistants/zBearer zapplication/json) Authorizationz Content-TypeN system_prompt)jsonheadersstatusri,z;push_to_telnyx: successfully updated assistant %s (HTTP %s)Tz**..!126 LLW  !?&wi0.  + "&"6"6z"BBM& 6Gyy$"&7G!TTx2OO!--"002--g&}}'7 - --!%---- f"s" Q   LLN  IC U----------*    H,   sA GF#E'$5FE*AFE,FF 8E.9F < E2 F E0F  F$F%)FGF&G'F*F,F.F 0F 2F 8E;9F F F FF FFG=G?GGc|jy |jjdS#t$rtj dYywxYw)z=Fetch last 3 conversation summaries from conversation_engine.zNo prior conversationsz$Failed to fetch recent conversations)rget_recent_summaryr?r6r@)rs rrz.SystemPromptInjector._get_recent_conversationssN  # # ++ ,++>>qA A ,   C D+ ,s*A  A c|jy |jjd}|yt|tr|j dSt |S#t $rtjdYywxYw)z)Fetch active Kinan directives from Redis.zNo active directiveszkinan:directives:activeutf-8z+Failed to fetch Kinan directives from Redis rr5 isinstancebytesdecodestrr?r6r@rvals rrz$SystemPromptInjector._get_directivessq :: ) ***..!:;C{-*4S%*@3::g& Nc#h N *   J K) *A A AA:9A:c|jy |jjd|}|yt|tr|j dSt |S#t $rtjd|YywxYw)z-Fetch caller info for the session from Redis.Unknownz aiva:state:rMz5Failed to fetch caller info from Redis for session %srN)rr$rTs rrz%SystemPromptInjector._get_caller_infosz ::  **..;zl!;=A>c|jy |jjd}|yt|tr|j dSt |S#t $rtjdYywxYw)z"Fetch active task list from Redis.Nonezaiva:tasks:activerMz%Failed to fetch open tasks from RedisrNrSs rr z$SystemPromptInjector._get_open_taskssq ::  **..!45C{*4S%*@3::g& Nc#h N    D E rUc|tjtj}|tz}|j dS)z6Return current AEST time (UTC+10) as formatted string.z%Y-%m-%d %H:%M:%S)rnowrutc AEST_OFFSETstrftime)utc_nowaest_nows rr!z#SystemPromptInjector._get_aest_time s2,,x||,[(  !455r)NNN)rz Optional[Any])r$rRreturnrR)r$rRrArRrabool)rarR) __name__ __module__ __qualname____doc__rr)rHrrrr staticmethodr!rrr r 4sV*!%)  #  # JWz, *  66rr )rf __future__rloggingr3rrrtypingrr getLoggerrcr6r]r8r"r rhrrrmsS2# 22   8 $b! -  Z6Z6r