՞iDdZddlmZddlZddlmZej eZ ddl Z ddl m Z dZ e d d d d ddZe dddd ddZe dddd ddZe d d d d d dZe dddd d!dZy#e $rdZ dZ YlwxYw)"u Genesis Task Queue — Dramatiq actor definitions. Each actor maps to a priority queue and has explicit retry + backoff settings. Workers are imported by the Dramatiq CLI worker process (``dramatiq core.task_queue.workers``). Queue assignments ----------------- critical voice/realtime webhooks — tight SLA, max_retries=1 high customer-facing operations — max_retries=5, fast back-off default background processing — max_retries=3, standard back-off low analytics / reporting — max_retries=5, long back-off (runs off-peak) Backoff values are in milliseconds (Dramatiq convention). When dramatiq is not installed the module still imports cleanly. Each actor function gets a no-op ``.send()`` and ``.send_with_options()`` stub so call sites don't need to guard against ImportError. # VERIFICATION_STAMP # Story: M6.02 — workers.py — Dramatiq actor definitions # Verified By: parallel-builder # Verified At: 2026-02-25T00:00:00Z # Tests: 8/8 # Coverage: 100% ) annotationsN)Any)actorTFc dd}|S)z?No-op ``@actor`` decorator used when dramatiq is not installed.c8dfd _fd_S)NcFtjdjy)Nubdramatiq not installed — task '%s' was not dispatched. Install with: pip install dramatiq[redis])loggerwarning__name__)argskwargsfns 0/mnt/e/genesis-system/core/task_queue/workers.py _noop_sendz._actor.._decorator.._noop_send1s@KKc|i|S)N)akrs rz,_actor.._decorator..9s:q3FA3Fr)r rr rreturnNone)sendsend_with_options)rrs`@r _decoratorz_actor.._decorator0s !BG#FB Ir)rrrrr)_kwargsrs r_actorr.s rdefaultii`) queue_name max_retries min_backoff max_backoffctjd|ddl}ddlm}|}|j |j y)ac Run a nightly epoch processing cycle. Delegates to ``NightlyEpochRunner.run_epoch()`` which aggregates the past 7 days of royal_conversations, distils axioms via Gemini Pro, and persists results to Qdrant + KG jsonl files. Parameters ---------- epoch_id : str Identifier for the epoch run (e.g. ``"2026-W09"``). z"Processing epoch task: epoch_id=%srN)NightlyEpochRunner)r infoasynciocore.epoch.nightly_epoch_runnerr%run run_epoch)epoch_idr'r%runners rprocess_epoch_taskr-Ds6 KK4h?B  !F KK  "#rhighii0uc2tjd||y)aB Dispatch a notification via the configured channel. Parameters ---------- notification_type : str One of ``"email"``, ``"sms"``, ``"webhook"``. recipient : str Destination address or phone number. data : dict Notification payload (subject, body, template vars, etc.). zSending %s notification to %sNr r&)notification_type recipientdatas rsend_notificationr5Zs KK'rcriticaldcz|jdxs|jdd}tjd|y)ay Handle an incoming Telnyx voice webhook with minimal latency. Uses ``queue_name="critical"`` so dedicated workers drain this queue before any other. ``max_retries=1`` avoids double-processing a call event. Parameters ---------- payload : dict Telnyx webhook JSON body. Must contain ``call_control_id`` or ``call_id`` at minimum. call_control_idcall_idunknownz$Processing voice webhook: call_id=%sN)getr r&)payloadr;s rprocess_voice_webhookr?ps2kk+,Q Iy0QG KK6@rc2tjd||y)a Write an entity update to the Genesis Knowledge Graph. Parameters ---------- entity_type : str KG entity category (e.g. ``"conversation"``, ``"axiom"``). entity_id : str Unique entity identifier. data : dict Entity fields to upsert. z&KG update: entity_type=%s entity_id=%sNr1) entity_type entity_idr4s rupdate_knowledge_graphrCs KK8+yQrlowiic2tjd||y)a Generate an analytics report during off-peak hours. Parameters ---------- report_type : str Report category (e.g. ``"weekly_revenue"``, ``"epoch_summary"``). period : str Reporting period string (e.g. ``"2026-W09"``). z5Generating analytics report: report_type=%s period=%sNr1) report_typeperiods rgenerate_analytics_reportrHs KK?r)r+strrr)r2rIr3rIr4dictrr)r>rJrr)rArIrBrIr4rJrr)rFrIrGrIrr)__doc__ __future__rloggingtypingr getLoggerr r dramatiqrrDRAMATIQ_AVAILABLE ImportErrorr-r5r?rCrHrrrrSs4#   8 $ (69!FS$T$*6qcvNO*:1#5QARA$9!FS RT R"5aUPQWs B BB