iLdZddlZddlZddlZddlmcmZddlm Z m Z ddl m Z m Z ddl mZ ddlZej$eZe GddZGdd Zy#e$rdZY5wxYw) u AIVA RLM Nexus — BinduHydrator Base Class Scatter/gather engine that fans out parallel tasks to fetch AIVA's context before she picks up a call. Story 4.01 — Track A File: core/hydrators/bindu_hydrator.py VERIFICATION_STAMP Story: 4.01 Verified By: parallel-builder Verified At: 2026-02-25 Tests: 24/24 Coverage: 100% Story 4.02 — Track A (extension) VERIFICATION_STAMP Story: 4.02 Verified By: parallel-builder Verified At: 2026-02-25 Tests: 6/6 Coverage: 100% Story 4.03 — Track A (extension) VERIFICATION_STAMP Story: 4.03 Verified By: parallel-builder Verified At: 2026-02-25 Tests: 11/11 Coverage: 100% Story 4.04 — Track A (extension) VERIFICATION_STAMP Story: 4.04 Verified By: parallel-builder Verified At: 2026-02-25 Tests: 9/9 Coverage: 100% Story 4.05 — Track A (extension) VERIFICATION_STAMP Story: 4.05 Verified By: parallel-builder Verified At: 2026-02-25 Tests: 10/10 Coverage: 100% N) dataclassfield)datetimetimezone)OptionalcVeZdZUdZeed<eed<eed<eed<dZeed<dZ eed<y ) HydrationSessionzs Tracks the state of a single call hydration cycle. One HydrationSession is created per incoming call. session_id aiva_call_id started_atstatusrtasks_dispatchedtasks_completedN) __name__ __module__ __qualname____doc__str__annotations__rrintr6/mnt/e/genesis-system/core/hydrators/bindu_hydrator.pyr r @s4O KcOSrr ceZdZdZdZdZdZddZdeded e fd Z ded e fd Z d e e fd Zd ed eefdZdd eded ee fdZdeded efdZy) BinduHydratoraP Scatter/gather engine: fans out N parallel Redis-backed tasks, aggregates into a single ROYAL_CHAMBER_CONTEXT XML envelope. This is the BASE class. Stories 4.02-4.07 will add fetch methods (memory, PG transcript history, Qdrant semantic recall, etc.). All hydration must complete within HYDRATION_DEADLINE_MS. izaiva:context:{session_id}i,Nc.||_||_||_y)z Args: redis_client: Async/sync redis client (must support setex). postgres_client: Optional asyncpg or psycopg2 connection/pool. qdrant_client: Optional Qdrant async client. N)redispgqdrant)self redis_clientpostgres_client qdrant_clients r__init__zBinduHydrator.__init__`s" !# rr call_idreturncKtjtj}t |||ddd}|j j |} |jj||jdd{tjd||j|S7'#t$r"}tjd||Yd}~|Sd}~wwxYww)aP Creates a HydrationSession and marks it 'pending' in Redis. Writes key aiva:context:{session_id} = "pending" with a 300s TTL using SETEX so the key auto-expires if the call never completes. Redis failure is NON-FATAL: the session is still returned so the call can proceed (degraded but live). Args: session_id: Unique identifier for this hydration cycle. call_id: Telnyx call_control_id (or internal call UUID). Returns: HydrationSession with status="pending" and started_at=UTC now. pendingr)r r r r rrr Nz8BinduHydrator: Redis key '%s' set to 'pending' (TTL=%ds)uOBinduHydrator: Redis SETEX failed for key '%s': %s — proceeding without cache)rnowrutcr _CONTEXT_KEY_TEMPLATEformatrsetex_CONTEXT_TTL_SECONDSloggerdebug Exceptionerror)r r r%now_utcsession redis_keyexcs rstart_hydrationzBinduHydrator.start_hydrationks",,x||,"!   ..555L  **""9d.G.GS S S LLJ))  T   LLa    sBAC*B(>B&?%B($C&B(( C1CCCCc|Kd|}d} tj|jj||jj|d{\}}i}|Q tj|}t|ts+t jdt|ji}g}|Q tj|}t|t s+t jdt|jg}||dS7#t$r%}t j d|igdcYd}~Sd}~wwxYw#tjtf$r"}t jd|i}Yd}~d}~wwxYw#tjtf$r"}t jd |g}Yd}~d}~wwxYww) u Fires 2 parallel Redis reads via asyncio.gather (L1 fetch — fastest tier). Keys read: 1. aiva:state:{session_id} → AIVA working state (JSON object) 2. kinan:directives:active → Kinan's current active directives (JSON array) Returns: { "aiva_state": dict — parsed JSON; {} if key missing or parse error "kinan_directives": list — parsed JSON; [] if key missing or parse error } Redis failure is NON-FATAL: returns safe defaults so the call can proceed without crashing the hydration pipeline. z aiva:state:zkinan:directives:activeNuRBinduHydrator._scatter_redis_tasks: Redis gather failed: %s — returning defaults aiva_statekinan_directivesuABinduHydrator: aiva:state is not a dict (%s) — defaulting to {}uGBinduHydrator: Failed to parse aiva:state JSON: %s — defaulting to {}uNBinduHydrator: kinan:directives:active is not a list (%s) — defaulting to []uTBinduHydrator: Failed to parse kinan:directives:active JSON: %s — defaulting to [])asynciogatherrgetr2r0r3jsonloads isinstancedictwarningtyperJSONDecodeError ValueErrorlist) r r aiva_keydirectives_keyaiva_rawdirectives_rawr7r;r<s r_scatter_redis_tasksz"BinduHydrator._scatter_redis_taskss"! -2 >-4^^ x( ~..( $Hn   !ZZ1 !*d3NN[Z(11"$J"$  % &#'::n#= !"2D9NNh-.77(*$)>NOO_( > LLd #%"= =  >$((*5 ]  $((*5 &j$&  &sF<A DD DF<#AE3F<8AE>F< D D=D82D=3F<8D==F<E;E61F<6E;;F<>F9F4/F<4F99F<cKdttffd }tj}|j d|d{S7w)u~ Fetches the most recent Kinan conversation from the royal_conversations table. SQL (parameterized): SELECT conversation_id, started_at, summary, decisions_made, action_items, kinan_directives, emotional_signal, key_facts FROM royal_conversations WHERE participants->>'kinan' = 'true' ORDER BY started_at DESC LIMIT 1 The postgres_client is expected to expose a psycopg2 connection pool with getconn()/putconn() methods. The sync DB call is wrapped in run_in_executor() so it does not block the event loop. Returns: dict — row from royal_conversations (RealDictCursor) when found. None — when no Kinan conversation exists, or on any DB error. DB failure is NON-FATAL: degraded context is better than a crashed call. r&cd} jj}|jtjj 5}|j dd|j}|( ddd|jj|yyt|cddd|jj|SS#1swYnxYwnJ#t$r>}tjd|Yd}~|jj|yyd}~wwxYw |jj|yy#|jj|wwxYw)u7Synchronous psycopg2 fetch — runs in executor thread.N)cursor_factorya? SELECT conversation_id, started_at, summary, decisions_made, action_items, kinan_directives, emotional_signal, key_facts FROM royal_conversations WHERE participants->>'kinan' = %s ORDER BY started_at DESC LIMIT 1 )trueuLBinduHydrator._scatter_postgres_task: DB fetch failed: %s — returning None) rgetconncursorpsycopg2extrasRealDictCursorexecutefetchoneputconnrCr2r0r3)conncurrowr7r s r _sync_fetchz9BinduHydrator._scatter_postgres_task.._sync_fetchsLD! *ww([[0N0N[O%SVKK "",,.C{#)%<#GGOOD)$ 9-%%<#GGOOD)$=%%%.  b#GGOOD)$ /%<#GGOOD)$4#GGOOD)$sTAC &C/C C CC CD? DD4D?DD?? EN)rrCr=get_event_looprun_in_executor)r r]loops` r_scatter_postgres_taskz$BinduHydrator._scatter_postgres_tasksA.$ *Xd^$ *L%%'))$ <<<Kd}d}|jtjdgS |j|}|jj ||||d{}g}|D]}t |dr|j r |j ni} |j| jddt|j| jd d| jd dd |S7#t$r"} tjd | gcYd} ~ Sd} ~ wwxYww) uI Semantic search in the aiva_conversations Qdrant collection (L3 fetch — semantic scars). Steps: 1. Embed query_text using _embed_query() (deterministic 768-dim vector for testing; in production this would be Gemini text-embedding-004). 2. Search the aiva_conversations collection with score_threshold=0.7. 3. Return up to top_k results whose score exceeds the threshold. Each result dict contains: - chunk_text (str) — the matched conversation chunk - score (float) — similarity score from Qdrant - conversation_id (str) — source conversation identifier - timestamp (str) — ISO-8601 timestamp of the chunk Returns: list[dict] — up to top_k results above the 0.7 threshold. [] — if no results above threshold, or on any exception (non-fatal). Args: query_text: Natural-language query to search for. top_k: Maximum number of results to return (default 3). gffffff?aiva_conversationsNuPBinduHydrator._scatter_qdrant_task: No Qdrant client configured — returning [])collection_name query_vectorlimitscore_thresholdpayload chunk_textconversation_id timestamp)rxscorerzr{uMBinduHydrator._scatter_qdrant_task: Qdrant search failed: %s — returning []) rr0rDrosearchhasattrrwrjr?floatr|r2r3) r rbrpSCORE_THRESHOLDCOLLECTION_NAMErt raw_resultsresultshitrwr7s r_scatter_qdrant_taskz"BinduHydrator._scatter_qdrant_taskAs 0. ;;  NNb I ,,Z8L $ 2 2 /) / !3!K#%G" )0i)@S[[#++VX&-kk,&C!&syy!1+2;;7H"+M%,[[b%A  N'*  LL_ I  sG(D3C/C-B C/,D-C// D8DDDDDc ,Kd}igd}d}g} tjtj|j||j |j |d|d{}|\}} } t |trtjd|n|}t | trtjd| n| }t | trtjd | n| }tj|jd i} tj|jdg} |tj|nd}tj|}d| d| d|d|d }tj ||j"j%|} |j&j)||j*|d{tj-d||j*|S7#tj$r&tjd t|d zYFt$r!} tjd | Yd} ~ jd} ~ wwxYw7#t$r"} tjd|| Yd} ~ |Sd} ~ wwxYww)a Fires all 3 scatter tasks in parallel, assembles a ROYAL_CHAMBER_CONTEXT XML envelope, caches it in Redis, and returns the XML string. Execution: 1. Fires _scatter_redis_tasks, _scatter_postgres_task, and _scatter_qdrant_task concurrently via asyncio.gather with a 450ms timeout (under the 500ms HYDRATION_DEADLINE_MS budget). 2. Any task that fails or times out is replaced with its safe default (aiva_state={}, kinan_directives=[], postgres_result=None, qdrant_results=[]) so the remaining results are still assembled. 3. Assembles a ROYAL_CHAMBER_CONTEXT XML envelope: {json} {json} {json or "none"} {json} 4. Writes the assembled XML to Redis key aiva:context:{session_id} with TTL 300s (SETEX). Redis failure is non-fatal. 5. Updates the HydrationSession status to "ready" (via start_hydration). Args: session_id: Unique identifier for this hydration cycle. call_id: Telnyx call_control_id (or internal call UUID). Returns: The assembled ROYAL_CHAMBER_CONTEXT XML string. g?r:NT)return_exceptions)timeoutuFgather_and_assemble: _scatter_redis_tasks raised %s — using defaultsuDgather_and_assemble: _scatter_postgres_task raised %s — using Noneu@gather_and_assemble: _scatter_qdrant_task raised %s — using []uWgather_and_assemble: gather timed out after %dms — proceeding with collected defaultsiuOgather_and_assemble: unexpected gather failure: %s — proceeding with defaultsr;r<nonez+z&z%z#z(r)zCgather_and_assemble: cached ROYAL_CHAMBER_CONTEXT at '%s' (TTL=%ds)uJgather_and_assemble: Redis SETEX failed for key '%s': %s — cache skipped)r=wait_forr>rMrarrBr2r0r3 TimeoutErrorrr@dumpsr?ET fromstringr,r-rr.r/r1)r r r%_GATHER_TIMEOUT_S redis_resultpostgres_resultqdrant_resultsoutcomes raw_redis raw_postgres raw_qdrantr7aiva_state_jsonking_directives_jsonlast_conversation_jsonrelated_scars_jsonxml_strr6s rgather_and_assemblez!BinduHydrator.gather_and_assembles<"-/BG *.!. $----j9//1--g6&*  *H3; /I|Z)Y/ \ ) , 2 Z  #/*i0 V ",**\%5%5lB%GH#zz,*:*:;Mr*RS+:+FDJJ 'F "ZZ7 ##2"34 456""8!9:012'  '  g..555L  **""9d.G.GQ Q Q LLU)) gH##  LLi%,-   LLa   : R   LL\    s JAH,G=-BH2B8J+*I&I$%I&;J=H5I!5J8I!IJI!!J$I&& J/J J JJ)NN))rrrrHYDRATION_DEADLINE_MSr,r/r$rr r8rCrMrrarHrrorrrrrrrrNs 8 $..c.FV.`DPSDPTDPL>=htn>=@stE{@>S>>TRVZ>@zCz#z#zrr)rr=r@loggingxml.etree.ElementTreeetree ElementTreer dataclassesrrrrtypingrpsycopg2.extrasrT ImportError getLoggerrr0r rrrrrs|.^ ""('   8 $     mm)HsAA)(A)