i ;|dZddlZddlZddlmZmZddlmZddlmZedZ dZ Gdd Z Gd d Z y) u Telnyx Webhook Interceptor — call lifecycle event handler. Story 3.02 — AIVA RLM Nexus PRD v2 — Track A Handles Telnyx call lifecycle events: - call.initiated → open a royal_conversations row (started_at = now, ended_at = NULL) Design principles (hardwired): - The phone call MUST NEVER break due to a DB or logging failure. - ALL exceptions from DB/IO operations are caught, logged, and swallowed. - session_id is extracted from the canonical Telnyx payload path; UUID4 is used as a fallback when the path is absent or malformed. - All timestamps are UTC. - The events log file (events.jsonl) uses the same path as ExecutionTelemetryInterceptor so a single file tracks all Genesis observability events. N)datetimetimezone)Path)Optionalz5/mnt/e/genesis-system/data/observability/events.jsonl)datapayloadcall_session_idceZdZdZddZdedefdZdedefdZdedefdZdede fd Z ded e de e fd Z d e ddfd Z d e ddfdZde deddfdZy)TelnyxWebhookInterceptora Handles Telnyx call lifecycle events. Responsibilities: - handle_call_initiated: records conversation start in royal_conversations - All DB failures are non-fatal (the call must never break) - All events are appended to the observability log Args: db_conn: Optional psycopg2 connection. When None, DB operations are skipped. redis_client: Optional Redis client. Reserved for future use (e.g. active-call cache). Nc ||_||_y)N)_db_redis)selfdb_conn redis_clients E/mnt/e/genesis-system/core/interceptors/telnyx_webhook_interceptor.py__init__z!TelnyxWebhookInterceptor.__init__3s" rreturncK|j|} |j||jdd|id|dS#t$r(}|jdt ||dYd}~Ed}~wwxYww)a5 Handle a Telnyx ``call.initiated`` webhook event. Processing steps: 1. Extract call_session_id from payload["data"]["payload"]["call_session_id"]. Falls back to a new UUID4 string if the key path is absent. 2. Insert a new row into royal_conversations with: - conversation_id = session_id - started_at = UTC now - ended_at = NULL (not yet finished) - participants = {"kinan": false, "aiva": false} Uses ON CONFLICT DO NOTHING for idempotency (duplicate webhooks are safe). 3. Append a ``telnyx_call_initiated`` event to the observability log. 4. Return {"status": "ok", "session_id": session_id}. Any DB or IO failure is caught, logged to the observability file (best-effort), and the method still returns {"status": "ok"} so the call is never disrupted. Args: payload: Parsed Telnyx webhook body (dict). Returns: {"status": "ok", "session_id": str} db_errorerror session_idNcall_initiatedrok)statusr)_extract_session_id_insert_conversation Exception _log_eventstr)rrrexcs rhandle_call_initiatedz.TelnyxWebhookInterceptor.handle_call_initiated;sz2--g6  W  % %j 1 (<*DEj99  W OOJ#c(*(U V V Ws+A3?A3 A0A+&A3+A00A3c`Kddl}|j|} |jrftjdt j tjjd}|jjd|| |j| t}|j!|j#|d d{|jdd|idddS#t$r(}|jdt||dYd}~d}~wwxYw#t$r(}|jdt||dYd}~d}~wwxYw7#|j$$r|jd |d d Yt$r(}|jd|t|d Yd}~d}~wwxYww)u Handle a Telnyx ``call.hangup`` webhook event. Processing steps: 1. Extracts session_id from payload["data"]["payload"]["call_session_id"]. Falls back to a UUID4 string if the path is absent or malformed. 2. Updates Redis key ``aiva:state:{session_id}`` to a JSON object: {"status": "ended", "ended_at": } The Redis write is NON-FATAL — if it fails the event is logged and execution continues. 3. Updates Postgres ``royal_conversations.ended_at`` to UTC now for the row whose conversation_id matches session_id. The Postgres write is NON-FATAL — if it fails the event is logged and execution continues. 4. Spawns PostCallEnricher.enrich(session_id) with a hard asyncio deadline of 3.0 seconds. - If it finishes within 3 s → enrichment_started = True (already done) - If it raises within 3 s → error logged, no crash - If it times out at 3 s → TimeoutError logged, no crash 5. Returns {"status": "ok", "enrichment_started": True} in all cases. ANY failure is logged and swallowed — the call must NEVER break. Args: payload: Parsed Telnyx webhook body (dict). Returns: {"status": "ok", "enrichment_started": True} rNended)rended_at aiva:state: redis_errorrrg@)timeoutenricher_timeoutz#PostCallEnricher timed out after 3s)rrenricher_error call_hanguprrT)renrichment_started)asynciorrjsondumpsrnowrutc isoformatsetr r!r"_update_conversation_endedPostCallEnricherwait_forenrich TimeoutError)rrr/rstater#enrichers rhandle_call_hangupz+TelnyxWebhookInterceptor.handle_call_hangupas< --g6  Z{{ % ( X\\ : D D F$ +j\ :EB  W  + +J 7 ')H""8??:#>"L L L  j'ABd;;7 Z OOMSXZ+X Y Y Z  W OOJ#c(*(U V V W M##  OO")4YZ   OO )CH=   sF.A2C+ D/E EEF.+ D4DF.DF. E(E F. EF.E$F+9F.;F+F&!F.&F++F.cK|j|}|j|d} |jrftjdt j tjjd}|jjd|||jd||xsd d d d |d S#t$r(}|jdt||dYd}~Kd}~wwxYww)u# Handle a Telnyx ``call.answered`` webhook event. Processing steps: 1. Extracts session_id from payload["data"]["payload"]["call_session_id"]. Falls back to a UUID4 string if the path is absent or malformed. 2. Extracts call_control_id from payload["data"]["payload"]["call_control_id"]. None is acceptable (non-fatal). 3. Sets Redis key ``aiva:state:{session_id}`` to a JSON object: {"status": "active", "answered_at": } The Redis write is NON-FATAL — if it fails the event is logged and execution continues. 4. Appends a ``call_answered`` event to the observability log. 5. Returns {"status": "ok", "capture_started": True, "session_id": session_id}. ANY failure is logged and swallowed — the call must NEVER break. Args: payload: Parsed Telnyx webhook body (dict). Returns: {"status": "ok", "capture_started": True, "session_id": str} call_control_idactive)r answered_atr(r)rN call_answeredunknown)rr?rT)rcapture_startedr)r_extract_fieldrr0r1rr2rr3r4r5r r!r")rrrr?r;r#s rhandle_call_answeredz-TelnyxWebhookInterceptor.handle_call_answereds0--g6 --g7HI Z{{ &#+<< #=#G#G#I$ +j\ :EB $.;)*  4zRR Z OOMSXZ+X Y Y Zs/$C,A2B8C,8 C)C$C,$C))C,c |}tD]}||} |rt|tr|Stt j S#ttf$rY.wxYw)a; Extract call_session_id from the canonical Telnyx payload path. Canonical path: payload["data"]["payload"]["call_session_id"] Returns: The session ID string from the payload, or a new UUID4 string if the path is absent, None, or the payload is not a dict. ) _SESSION_PATH isinstancer"KeyError TypeErroruuiduuid4)rrnodekeys rrz,TelnyxWebhookInterceptor._extract_session_idsa D$ !Cy ! 4- 4::<  )$   s%AAAfieldcP |dd|}||SdS#ttf$rYywxYw)a Extract an arbitrary field from the canonical Telnyx payload path ``payload["data"]["payload"][field]``. Args: payload: Parsed Telnyx webhook body. field: Field name to look up inside ``data.payload``. Returns: The field value as a string, or None if the path is absent, None-valued, or the payload structure is unexpected. rrN)rJrK)rrrPvalues rrEz'TelnyxWebhookInterceptor._extract_fieldsA FOI.u5E!-5 74 7)$  s%%rcv|jsyd}tjtj}t j ddd}|jj} |j||||f|jj|jy#|jwxYw)a Insert a new row into royal_conversations for this call session. Schema (from data/migrations/001_royal_conversations.sql): - conversation_id UUID PRIMARY KEY - started_at TIMESTAMP NOT NULL - ended_at TIMESTAMP (NULL until hangup) - participants JSONB NOT NULL DEFAULT '{"kinan": false, "aiva": false}' Uses ON CONFLICT DO NOTHING so duplicate call.initiated events are safe. Args: session_id: UUID string for this call session. Raises: Any psycopg2 exception if the DB operation fails. Callers are responsible for catching and swallowing. Nz INSERT INTO royal_conversations (conversation_id, started_at, participants) VALUES (%s, %s, %s) ON CONFLICT (conversation_id) DO NOTHING F)kinanaiva) r rr2rr3r0r1cursorexecutecommitclose)rrqueryr2 participantsrVs rrz-TelnyxWebhookInterceptor._insert_conversations&xx  ll8<<(zzE5"AB "  NN5:sL"A B HHOO  LLNFLLNs &/B&&B8cD|jsyd}tjtj}|jj } |j |||f|jj|jy#|jwxYw)a3 Set ended_at = UTC now for royal_conversations row matching session_id. Args: session_id: UUID string for this call session. Raises: Any psycopg2 exception if the DB operation fails. Callers are responsible for catching and swallowing. Nzu UPDATE royal_conversations SET ended_at = %s WHERE conversation_id = %s ) r rr2rr3rVrWrXrY)rrrZr2rVs rr6z3TelnyxWebhookInterceptor._update_conversation_ended)srxx   ll8<<("  NN53 "3 4 HHOO  LLNFLLNs .B B event_typercp tjjddtjt j jd|d|}ttd5}|jtj|dzdddy#1swYyxYw#t$rYywxYw)a Append an event record to the observability log (best-effort, never raises). Format: one JSON object per line, compatible with events.jsonl used by ExecutionTelemetryInterceptor. Args: event_type: Short string label (e.g. "call_initiated", "db_error"). data: Additional key/value pairs to include in the event record. T)parentsexist_oktelnyx_) timestampr]a N) EVENTS_LOG_PATHparentmkdirrr2rr3r4openwriter0r1r )rr]reventfhs rr!z#TelnyxWebhookInterceptor._log_eventEs   " " ( ( ( E%\\(,,7AAC ' |4E os+ 3rE*T12 3 3 3   s0A*B),(BB)B&"B)&B)) B54B5)NN)__name__ __module__ __qualname____doc__rdictr$r=rFr"rrrErr6r!rrr r %s #$:4$:D$:LE<E<ErvsF  ' NO7 uux r