i~VdZddlZddlZddlZddlZddlZddlZddlmZm Z eje Z Gdde ZGddZy)ub Story 3.08: PostCallEnricher — Gemini Enrichment Story 3.09: PostCallEnricher — Qdrant Write + Deadline Handler Story 3.10: PostCallEnricher — Postgres Write + Redis Cleanup AIVA RLM Nexus PRD v2 — Track A Story 3.08: Reads a call transcript from Redis (LRANGE aiva:transcript:{session_id}), sends it to Gemini 2.0 Flash for enrichment, and returns a parsed 7-field enrichment object. Story 3.09: Adds the public enrich() entry point which orchestrates: 1. Gemini enrichment (Story 3.08) 2. 768-dim deterministic embedding of the summary 3. Qdrant upsert into "aiva_conversations" collection Returns the Qdrant vector UUID on success, None on any failure (non-fatal). Story 3.10: Extends enrich() with: 4. Postgres UPDATE on royal_conversations with all enriched fields and memory_vector_id (from Qdrant) 5. Redis DELETE of aiva:transcript:{session_id} ONLY after successful Postgres write (preserves data for retry on Postgres failure) ALL external I/O (Redis, Gemini API, Qdrant, Postgres) is injected — zero hardwired side effects. N)AnyOptionalceZdZdZy)EnrichmentErrorz@Raised when Gemini enrichment fails and the fallback also fails.N)__name__ __module__ __qualname____doc__:/mnt/e/genesis-system/core/enrichers/post_call_enricher.pyrr0sJr rc *eZdZdZdZdZdZ d#dededed ed df d Z d ed e efd Z d ed e fdZ ded efdZded efdZded efdZded e fdZde d e fdZded efdZdedede d efdZd ede de ed efdZd ed dfd Ze d$d!ed e fd"Zy)%PostCallEnricheraX Enriches a call transcript using Gemini 2.0 Flash. Reads transcript chunks from Redis, sends to Gemini, parses response. Designed for use after a Telnyx voice call ends. Usage: enricher = PostCallEnricher(redis_client=redis, gemini_api_key=api_key) result = await enricher._enrich_with_gemini(session_id) zgemini-2.0-flashsummaryentitiesdecisions_made action_itemsemotional_signal key_factskinan_directivesaiva_conversationsN redis_clientgemini_api_key qdrant_client postgres_poolreturnc<||_||_||_||_y)a# Args: redis_client: An async Redis client with lrange() and delete() support. gemini_api_key: Google AI Studio API key for Gemini calls. qdrant_client: A qdrant_client.QdrantClient instance (or mock). If None, Qdrant writes are skipped. postgres_pool: A psycopg2 connection pool (ThreadedConnectionPool or SimpleConnectionPool) with getconn()/putconn() methods. If None, Postgres writes are skipped. N)_redis_gemini_api_key_qdrant_postgres_pool)selfrrrrs r __init__zPostCallEnricher.__init__Os""# -$ +r session_idcK |j|d{}|j dd}||j dk(s|stj d|yttj}|j|}t|}||d<||d<|j|||d{}|stjd|y|j|||d{} | r|j|d{|Stjd ||S7 #t$r!}tjd||Yd}~yd}~wwxYw77k7Rw) u Main entry point called by WebhookInterceptor on call.hangup. Must complete within 3 seconds. Steps: 1. _enrich_with_gemini(session_id) — reads transcript, calls Gemini 2. Embed enriched["summary"] into a 768-dim vector 3. Upsert the enrichment payload to Qdrant aiva_conversations 4. Persist enriched fields + memory_vector_id to Postgres royal_conversations 5. Delete Redis transcript key (only on successful Postgres write) 6. Return the Qdrant vector UUID string, or None on any failure Args: session_id: Telnyx/AIVA call session identifier. Returns: UUID string (Qdrant vector_id) on success, None on failure. Nz=PostCallEnricher.enrich: enrichment failed for session %s: %sruHPostCallEnricher.enrich: skipping Qdrant write — empty call session %sr% vector_iduIPostCallEnricher.enrich: Qdrant write failed for session %s — non-fatalufPostCallEnricher.enrich: Postgres write failed for session %s — Redis transcript preserved for retry)_enrich_with_geminirloggererrorget_empty_enrichmentinfostruuiduuid4 _embed_textdict_write_to_qdrantwarning_persist_to_postgres_cleanup_redis) r#r%enrichedexcrr(vectorpayloadsuccesspg_oks r enrichzPostCallEnricher.enrichisi& !55jAAH,,y"- d,,.y9 9 KKZ  % !!'*x. * ( --iII NN[ // HiPP %%j1 1 1 NN7  ]B  LLO    0JQ 2snED(D%D(BE7E83E+E,EEE%D(( E1E E EEEEcKd|}|jj|ddd{}|s&tjd||j S|j |} |j |d{}|j|S7h7#t$r.}tjd||td|d||d}~wwxYww) ug Full enrichment pipeline for a finished call. Steps: 1. LRANGE aiva:transcript:{session_id} 0 -1 from Redis 2. Join chunks into a formatted transcript string 3. Call Gemini 2.0 Flash with structured enrichment prompt 4. Parse JSON response into the 7-field enrichment dict 5. Return enrichment dict Returns a dict with exactly these keys: summary (str) entities (list[str]) decisions_made (list[str]) action_items (list[dict]) — each: {"task", "owner", "deadline"} emotional_signal (str) key_facts (list[str]) kinan_directives (list[str]) On empty transcript: returns graceful empty enrichment. Raises EnrichmentError if Gemini fails AND the fallback also fails. aiva:transcript:rNz1PostCallEnricher: empty transcript for session %sz7PostCallEnricher: Gemini call failed for session %s: %sz%Gemini enrichment failed for session z: ) rlranger*r.r-_build_transcript_string _call_gemini Exceptionr+r_parse_gemini_response)r#r% redis_key raw_chunks transcript raw_responser9s r r)z$PostCallEnricher._enrich_with_geminis.'zl3 ;;--iB??  KKKZ X))+ +22:>  !%!2!2:!>>L**<88)@?  LLI  "7 |2cUK   sE&CB=C'B;B<BCB C )CC  CrHcg}|D]|}t|ttfr|jdd} t j |}|j dd}|j dd}|jd|d |~d j|S#tjtf$r|jt|YwxYw) z Decodes Redis LRANGE entries and joins them into a readable transcript. Each chunk is a JSON string with at least {"speaker": ..., "text": ...}. Unknown formats are included as-is so no data is silently dropped. utf-8replace)errorsspeakerUNKNOWNtextr'[z]  ) isinstancebytes bytearraydecodejsonloadsr,appendJSONDecodeErrorAttributeErrorr/join)r#rHlinesentrychunkrOrQs r rCz)PostCallEnricher._build_transcript_strings )E%%!34 WY ? ) 5)))Iy9yy, q D623 )yy((.9 ) SZ( )sAB3C  C rIcd|dS)z3Builds the structured prompt for Gemini enrichment.ugYou are an expert call analyst for an AI voice agent system. Analyse the following call transcript and return a JSON object with EXACTLY these 7 fields: { "summary": "<1-3 sentence summary of the call>", "entities": [""], "decisions_made": [""], "action_items": [ {"task": "", "owner": ", "deadline": ""} ], "emotional_signal": "", "key_facts": [""], "kinan_directives": [""] } Rules: - Return ONLY the raw JSON object — no markdown, no code fences. - If a field has no relevant content, use an empty list [] or an empty string. - action_items must always have task, owner, and deadline keys. TRANSCRIPT: --- z ---r )r#rIs r _build_enrichment_promptz)PostCallEnricher._build_enrichment_prompts (l)  r cK|j|} ddlm}|j|j|j |j }|j|}|jS#t$rYnwxYwddl }ddl }d|j d|j}tjdd|igigdd id jd }|jj!||d d id } |jj#| d5} tj$| j'j)d } dddn #1swYnxYw ddddddSw)a[ Makes the actual Gemini API call using the REST endpoint. Uses google.generativeai if available, otherwise falls back to urllib-based REST call so there is no hard dependency on the SDK. Returns the raw text response from Gemini. Raises any exception from the API layer for the caller to handle. rN)api_keyz8https://generativelanguage.googleapis.com/v1beta/models/z:generateContent?key=partsrQresponseMimeTypezapplication/json)contentsgenerationConfigrLz Content-TypePOST)dataheadersmethod)timeout candidatescontent)rbgoogle.generativeai generativeai configurer GenerativeModel GEMINI_MODELgenerate_contentrQ ImportErrorurllib.request urllib.errorrXdumpsencoderequestRequesturlopenrYreadrW) r#rIpromptgenaimodelresponseurlliburlr;reqrespbodys r rDzPostCallEnricher._call_geminis..z:  / OOD$8$8O 9))$*;*;._list_of_strZs+#t$.12dD 22I3s,c g}t|ts|S|D]s}t|ts|jt |j ddt |j ddt |j dddu|S)Ntaskr'ownerdeadline)rrr)rTrr3rZr/r,)rresultrs r _list_of_action_itemszEPostCallEnricher._normalise_enrichment.._list_of_action_items_sFc4(  dD)MM$'(<$=%('2)>%?(+DHHZ,D(E Mr rr'rrrrneutralrrr)rrr/r,)r#rrrs r rz&PostCallEnricher._normalise_enrichmentUs  c d  s t  6::i45$VZZ %;<*6::6F+GH1&**^2LM #FJJ/A9$M N%fjj&=> ,VZZ8J-K L  r rQc &tj|jj}d}t j |t |z }||zd|}td|dDcgc]!}tjd|||dzd#}}|Dcgc]}t j|r|nd}}td|Dd xsd} t j| r| dk(rd} |dd Dcgc]}|| z  c}Scc}wcc}wcc}w) a Creates a 768-dim deterministic embedding vector from text. Uses a hash-based approach (SHA-256 repeated) for testability and zero external dependencies. In production, swap the body with a real Gemini text-embedding-004 call without changing the interface. Args: text: The text to embed (typically the call summary). Returns: List of 768 floats normalised to [-1, 1]. i Nrfgc32K|]}t|ywr)abs).0vs r z/PostCallEnricher._embed_text..s.!s1v.sg?)defaulti) hashlibsha256r{digestmathceillenrangestructunpackisfinitemax) r#rQh needed_bytes repeat_countextendedivaluesrmax_vals r r2zPostCallEnricher._embed_textxs NN4;;= ) 0 0 2 yyA!67  $m|41lA.  MM#xAE2 3A 6  ;AAQt}}Q'!S0AA.v.<C}}W%CG%+DS\2G 22 B3s,&D D 5 Dr(r:r;cZK|jtjdy ddlm}|jj |j ||||gtjd||j y #t$r }tjd |Yd}~yd}~wwxYww) a Upserts a single point into the aiva_conversations Qdrant collection. Non-fatal: all exceptions are caught and logged; caller receives False. Args: vector_id: UUID string to use as the Qdrant point ID. vector: 768-dim float list. payload: Dict attached as Qdrant point payload (enrichment fields). Returns: True on successful upsert, False on any failure. NuIPostCallEnricher._write_to_qdrant: no Qdrant client injected — skippingFr) PointStruct)idr:r;)collection_namepointsz4PostCallEnricher._write_to_qdrant: upserted %s to %sTuCPostCallEnricher._write_to_qdrant: upsert failed (%s) — non-fatal) r!r*r5qdrant_client.modelsrupsert_QDRANT_COLLECTIONr.rEr+)r#r(r:r;rr9s r r4z!PostCallEnricher._write_to_qdrants <<  NN[   8 LL   $ 7 7#yQR   KKF''    LLU   s/#B+AA?>B+? B(B#B+#B((B+r8c ^K|jtjdyd}|jddt j |jdgt j |jdgt j |jd g|jd d t j |jd gt j |jd g||f }d} |jj }|j5}|j||ddd|jtjd|| ||jj|yy#1swYPxYw#t$rb}tjd|||! |jn#t$rYnwxYwYd}~||jj|yyd}~wwxYw#||jj|wwxYww)u Updates the royal_conversations row for session_id in Postgres. Columns written: transcript_raw — raw enriched summary string enriched_entities — JSON-serialised list of entity strings decisions_made — JSON-serialised list of decision strings action_items — JSON-serialised list of action-item dicts emotional_signal — string signal (positive/negative/neutral/…) key_facts — JSON-serialised list of fact strings kinan_directives — JSON-serialised list of directive strings memory_vector_id — Qdrant UUID returned from Story 3.09 Uses parameterized SQL (%s placeholders — NO string formatting in SQL). Uses getconn/putconn in a try/finally block. Returns: True on successful UPDATE, False on any failure or if no pool injected. NuMPostCallEnricher._persist_to_postgres: no Postgres pool injected — skippingFzUPDATE royal_conversations SET transcript_raw = %s, enriched_entities = %s, decisions_made = %s, action_items = %s, emotional_signal = %s, key_facts = %s, kinan_directives = %s, memory_vector_id = %s WHERE session_id = %srr'rrrrrrrzHPostCallEnricher._persist_to_postgres: updated session %s (vector_id=%s)TzGPostCallEnricher._persist_to_postgres: UPDATE failed for session %s: %s)r"r*r5r,rXrzgetconncursorexecutecommitr.putconnrEr+rollback) r#r%r8r(_SQLparamsconncurr9s r r6z%PostCallEnricher._persist_to_postgress2    & NN_  $  LLB ' JJx||J3 4 JJx||$4b9 : JJx||NB7 8 LL+Y 7 JJx||K4 5 JJx||$6; <     2&&..0D *# D&) * KKM KKZ  ##++D1 - * *  LLY  MMO ##++D1  ##++D1 s~DH-*F.F/F1H-FF H%H?GH GHGHH #H-HH H**H-cKd|} |jj|d{tjd|y7#t$r!}tj d||Yd}~yd}~wwxYww)z Deletes aiva:transcript:{session_id} from Redis after successful Postgres write. Failure is non-fatal: logged as a warning, exception not propagated. r@Nz/PostCallEnricher._cleanup_redis: deleted key %szA2AAAA2A A/A*%A2*A//A2rc|gggdggdS)z:Returns a valid enrichment dict with empty/default fields.rrr rs r r-z"PostCallEnricher._empty_enrichment+s#   ) "  r )r'NN)u%Empty call — no transcript captured)rrr r ru_ENRICHMENT_FIELDSrrr/r$rr>r3r)rrCrbrDrFrr2boolr4r6r7 staticmethodr-r r r rr5s &L. !!! ,,, ,  ,  ,4BsBx}BP,9C,9D,9d 4 C * 3 3 6-DS-DS-D^@3@4@&! D! T! F333<((&*(59( (\R2R2R2C= R2  R2hst&>        r r)r rrXloggingrrr0typingrr getLoggerrr*rErrr r r rsK8    8 $ i C C r