""" core/epoch/nightly_epoch_runner.py Story 9.02 — NightlyEpochRunner: Conversation Aggregator Story 9.03 — NightlyEpochRunner: Gemini Distillation Story 9.04 — NightlyEpochRunner: Axiom Write to KG Story 9.05 — NightlyEpochRunner: Full Orchestration + Epoch Log Queries royal_conversations from the past 7 days (Kinan conversations only) for weekly epoch distillation, then distils those conversations into axioms via Gemini Pro, then persists those axioms to Qdrant and the KG jsonl files. Story 9.05 adds run_epoch() — the full Sunday night pipeline — which runs all three stages in sequence and writes an epoch log entry. # VERIFICATION_STAMP # Story: 9.02 # Verified By: parallel-builder # Verified At: 2026-02-25T00:00:00Z # Tests: 9/9 # Coverage: 100% # VERIFICATION_STAMP # Story: 9.03 # Verified By: parallel-builder # Verified At: 2026-02-25T00:00:00Z # Tests: 8/8 # Coverage: 100% # VERIFICATION_STAMP # Story: 9.04 # Verified By: parallel-builder # Verified At: 2026-02-25T00:00:00Z # Tests: 41/41 # Coverage: 100% # VERIFICATION_STAMP # Story: 9.05 # Verified By: parallel-builder # Verified At: 2026-02-25T00:00:00Z # Tests: 36/36 # Coverage: 100% """ from __future__ import annotations import json import logging import os import time from datetime import datetime, timezone from typing import Any logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # SQL — uses Postgres INTERVAL so the 7-day window is evaluated server-side. # Python datetime arithmetic is intentionally NOT used in the query. # --------------------------------------------------------------------------- AGGREGATION_QUERY = """ SELECT conversation_id, started_at, transcript_raw, enriched_entities, decisions_made, action_items, key_facts, kinan_directives FROM royal_conversations WHERE started_at >= NOW() - INTERVAL '7 days' AND participants->>'kinan' = 'true' ORDER BY started_at ASC """ # Ordered column names matching the SELECT list above. _COLUMNS = ( "conversation_id", "started_at", "transcript_raw", "enriched_entities", "decisions_made", "action_items", "key_facts", "kinan_directives", ) # --------------------------------------------------------------------------- # Distillation prompt — sent to Gemini Pro with the week's conversations. # --------------------------------------------------------------------------- DISTILLATION_PROMPT = """ You are the Genesis Memory Distiller. Review this week's AIVA<>Kinan conversations. Extract 5-10 high-value axioms: standing patterns, recurring preferences, key facts about the business, and strategic decisions made. CONVERSATIONS: {conversations_json} Return ONLY valid JSON: {{ "axioms": [ {{"id": "epoch_{date}_001", "content": "...", "category": "preference|fact|strategy|directive", "confidence": 0.9}} ], "week_summary": "2-sentence summary of this week" }} """ # Required keys every axiom dict must contain. _AXIOM_REQUIRED_KEYS = frozenset({"id", "content", "category", "confidence"}) # --------------------------------------------------------------------------- # KG persistence paths and Qdrant collection name (Story 9.04) # --------------------------------------------------------------------------- KG_AXIOM_PATH = ( "/mnt/e/genesis-system/KNOWLEDGE_GRAPH/axioms/genesis_evolution_learnings.jsonl" ) WEEKLY_SUMMARY_PATH = ( "/mnt/e/genesis-system/KNOWLEDGE_GRAPH/entities/weekly_summaries.jsonl" ) QDRANT_COLLECTION = "genesis_axioms" # Epoch observability log — one JSON line per Sunday night run. EPOCH_LOG_PATH = "/mnt/e/genesis-system/data/observability/epoch_log.jsonl" class NightlyEpochRunner: """ Weekly epoch runner that aggregates, distills, and persists Genesis learnings. Stories implemented: 9.02 — aggregate_week(): query last 7 days of Kinan conversations. 9.03 — distill(): send conversations to Gemini Pro for axiom distillation. 9.04 — write_axioms(): embed and persist axioms to Qdrant + KG jsonl files. 9.05 — run_epoch(): full Sunday night pipeline with epoch log. Parameters ---------- pg_conn: Postgres connection (psycopg2-style) with a ``.cursor()`` method. Pass ``None`` for graceful degradation — all methods return safe empty values rather than raising. gemini_client: Async Gemini client with a ``generate(prompt, model)`` coroutine and an ``embed(text, model)`` coroutine. Pass ``None`` to skip distillation and embedding (returns safe values). qdrant_client: Qdrant client with an ``upsert(collection_name, points)`` method. Pass ``None`` to skip Qdrant upserts — file writes still occur. """ def __init__( self, pg_conn: Any = None, gemini_client: Any = None, qdrant_client: Any = None, ) -> None: self.pg = pg_conn self.gemini = gemini_client self.qdrant = qdrant_client # ------------------------------------------------------------------ # Public API # ------------------------------------------------------------------ async def aggregate_week(self) -> list[dict]: """ Query the last 7 days of Kinan<>AIVA conversations from Postgres. The 7-day window is computed by Postgres (``INTERVAL '7 days'``), not by Python datetime arithmetic. Only conversations that include Kinan as a participant (``participants->>'kinan' = 'true'``) are returned, ordered oldest-first. Returns ------- list[dict] Each dict has exactly 8 keys:: conversation_id — str UUID started_at — datetime (from Postgres) transcript_raw — str enriched_entities — dict | None decisions_made — list | None action_items — list | None key_facts — list | None kinan_directives — list | None Returns ``[]`` (empty list, never ``None``) when: * ``pg_conn`` is ``None`` (no database configured) * The query returns zero rows * Any database error occurs (logged at ERROR level) """ if self.pg is None: logger.debug( "NightlyEpochRunner.aggregate_week: pg_conn is None — " "returning empty list (graceful degradation)" ) return [] try: cursor = self.pg.cursor() cursor.execute(AGGREGATION_QUERY) rows = cursor.fetchall() cursor.close() except Exception: logger.exception( "NightlyEpochRunner: failed to aggregate week's conversations" ) return [] return [dict(zip(_COLUMNS, row)) for row in rows] # ------------------------------------------------------------------ # Story 9.03 — Gemini distillation # ------------------------------------------------------------------ async def distill(self, conversations: list[dict]) -> dict: """ Send conversations to Gemini Pro for axiom distillation. Formats the ``DISTILLATION_PROMPT`` with a JSON-serialised conversation list and asks the configured Gemini client (``gemini-pro`` model) to return structured axioms and a week summary. Parameters ---------- conversations: List of conversation dicts, typically the output of ``aggregate_week()``. Serialised via ``json.dumps`` before being embedded in the prompt. Returns ------- dict Always returns a dict with two keys:: "axioms": list[dict] — each has id, content, category, confidence "week_summary": str Graceful-degradation paths (all return safe empty values): * Empty ``conversations`` list → ``{"axioms": [], "week_summary": "No conversations this week"}`` * ``gemini_client`` is ``None`` → ``{"axioms": [], "week_summary": "No Gemini client configured"}`` * Gemini returns non-JSON or malformed JSON → ``{"axioms": [], "week_summary": "Distillation failed"}`` * Axioms missing required keys (id, content, category, confidence) are silently filtered out. """ if not conversations: return {"axioms": [], "week_summary": "No conversations this week"} if self.gemini is None: return {"axioms": [], "week_summary": "No Gemini client configured"} date_str = datetime.now(timezone.utc).strftime("%Y_%m_%d") prompt = DISTILLATION_PROMPT.format( conversations_json=json.dumps(conversations, default=str), date=date_str, ) try: raw = await self.gemini.generate(prompt, model="gemini-pro") result = json.loads(raw) # Validate and filter axioms — only keep dicts with all required keys. raw_axioms = result.get("axioms", []) valid_axioms = [ ax for ax in raw_axioms if isinstance(ax, dict) and _AXIOM_REQUIRED_KEYS.issubset(ax.keys()) ] return { "axioms": valid_axioms, "week_summary": result.get("week_summary", ""), } except (json.JSONDecodeError, Exception): logger.exception("NightlyEpochRunner: Gemini distillation failed") return {"axioms": [], "week_summary": "Distillation failed"} # ------------------------------------------------------------------ # Story 9.04 — KG persistence # ------------------------------------------------------------------ async def write_axioms(self, axioms: list[dict], week_summary: str) -> int: """ Persist axioms to Qdrant and the Genesis Knowledge Graph jsonl files. For each axiom: 1. Embed the axiom's ``content`` field with Gemini ``text-embedding-004`` (skipped when ``gemini_client`` is None). 2. Upsert the axiom as a ``PointStruct`` into the Qdrant ``genesis_axioms`` collection (skipped when ``qdrant_client`` or embedding is unavailable). 3. Append the axiom as a JSON line to ``KNOWLEDGE_GRAPH/axioms/genesis_evolution_learnings.jsonl`` (always attempted). After processing all axioms, writes ``week_summary`` as a JSON line ``{"date": "", "summary": ""}`` to ``KNOWLEDGE_GRAPH/entities/weekly_summaries.jsonl``. Partial-success model ~~~~~~~~~~~~~~~~~~~~~ If a single axiom's embedding or Qdrant upsert fails, the error is logged and processing continues with the next axiom. File-write failures for individual axioms are also logged and skipped. The method never raises — it returns whatever count it managed to write. Parameters ---------- axioms: List of axiom dicts (each with at minimum ``id`` and ``content``). week_summary: 2-sentence human-readable summary of the week's learnings. Returns ------- int Number of axioms successfully written to at least the jsonl file. Returns 0 on empty input without modifying any axiom file (the weekly summary is still written). """ # Ensure parent directories exist. os.makedirs(os.path.dirname(KG_AXIOM_PATH), exist_ok=True) os.makedirs(os.path.dirname(WEEKLY_SUMMARY_PATH), exist_ok=True) written = 0 for axiom in axioms: try: # Step 1: embed (if gemini available) vector: list[float] | None = None if self.gemini is not None: try: vector = await self.gemini.embed( axiom.get("content", ""), model="text-embedding-004", ) except Exception: logger.exception( "NightlyEpochRunner.write_axioms: embed failed for " "axiom id=%r — skipping Qdrant upsert", axiom.get("id"), ) vector = None # Step 2: upsert to Qdrant (if both client and vector available) if self.qdrant is not None and vector is not None: try: from qdrant_client.models import PointStruct self.qdrant.upsert( collection_name=QDRANT_COLLECTION, points=[ PointStruct( id=axiom["id"], vector=vector, payload=axiom, ) ], ) except Exception: logger.exception( "NightlyEpochRunner.write_axioms: Qdrant upsert failed " "for axiom id=%r — continuing", axiom.get("id"), ) # Step 3: append to KG jsonl (always attempted) with open(KG_AXIOM_PATH, "a", encoding="utf-8") as fh: fh.write(json.dumps(axiom) + "\n") written += 1 except Exception: logger.exception( "NightlyEpochRunner.write_axioms: failed to write axiom id=%r", axiom.get("id"), ) # Write week summary regardless of axiom outcomes. try: date_str = datetime.now(timezone.utc).strftime("%Y-%m-%d") summary_entry = {"date": date_str, "summary": week_summary} with open(WEEKLY_SUMMARY_PATH, "a", encoding="utf-8") as fh: fh.write(json.dumps(summary_entry) + "\n") except Exception: logger.exception( "NightlyEpochRunner.write_axioms: failed to write weekly summary" ) return written # ------------------------------------------------------------------ # Story 9.05 — Full Sunday night pipeline # ------------------------------------------------------------------ async def run_epoch(self) -> dict: """ Full Sunday night pipeline: aggregate → distill → write axioms → log. Stages ------ 1. ``aggregate_week()`` — query last 7 days of Kinan conversations. 2. ``distill(conversations)`` — Gemini Pro axiom distillation. 3. ``write_axioms(axioms, week_summary)`` — persist to Qdrant + KG files. 4. Write an epoch log entry to ``data/observability/epoch_log.jsonl``. Returns ------- dict Epoch log entry with keys: ``date`` ISO date string (``YYYY-MM-DD``) of the run. ``conversations_processed`` Number of conversations returned by ``aggregate_week()``. ``axioms_written`` Number of axioms successfully written by ``write_axioms()``. ``week_summary`` 2-sentence summary produced by ``distill()``. ``duration_s`` Wall-clock duration in seconds (float, 3 decimal places), measured with ``time.monotonic()`` for accuracy. ``status`` ``"success"`` if all three stages completed without exception, ``"failed"`` if any stage raised. ``error`` *(only on failure)* String representation of the exception that caused the failure. Error handling -------------- If ANY stage raises an exception, execution stops, the exception is logged (with traceback), ``status`` is set to ``"failed"``, and an epoch log entry is still written before returning. The method never re-raises the stage exception. """ start = time.monotonic() status = "success" conversations_processed = 0 axioms_written = 0 week_summary = "" error_msg = None try: # Stage 1: aggregate conversations = await self.aggregate_week() conversations_processed = len(conversations) # Stage 2: distill result = await self.distill(conversations) axioms = result.get("axioms", []) week_summary = result.get("week_summary", "") # Stage 3: write axioms axioms_written = await self.write_axioms(axioms, week_summary) except Exception as exc: logger.exception("NightlyEpochRunner.run_epoch: pipeline failed") status = "failed" error_msg = str(exc) duration = time.monotonic() - start epoch_log: dict = { "date": datetime.now(timezone.utc).strftime("%Y-%m-%d"), "conversations_processed": conversations_processed, "axioms_written": axioms_written, "week_summary": week_summary, "duration_s": round(duration, 3), "status": status, } if error_msg is not None: epoch_log["error"] = error_msg # Persist the epoch log entry (append, never overwrite). try: os.makedirs(os.path.dirname(EPOCH_LOG_PATH), exist_ok=True) with open(EPOCH_LOG_PATH, "a", encoding="utf-8") as fh: fh.write(json.dumps(epoch_log) + "\n") except OSError: logger.exception("NightlyEpochRunner: failed to write epoch log") return epoch_log