""" AIVA RLM Nexus PRD v2 — Story 4.07 Track A: ConversationReplayEngine — Last N Conversations Summary Queries royal_conversations, formats N most recent Kinan<>AIVA conversations into a concise human-readable string for system prompt injection. File: core/memory/conversation_replay.py """ import json from typing import Optional # --------------------------------------------------------------------------- # ConversationReplayEngine # --------------------------------------------------------------------------- class ConversationReplayEngine: """ Queries royal_conversations, formats N most recent conversations into a concise summary for system prompt injection. Uses the getconn/putconn pattern against a psycopg2 connection pool. All SQL uses parameterised %s placeholders — never f-strings in SQL. """ def __init__(self, postgres_pool=None): """ Args: postgres_pool: A psycopg2 connection pool (SimpleConnectionPool or ThreadedConnectionPool). If None, get_recent_summary() returns an empty string gracefully. """ self._pool = postgres_pool # ----------------------------------------------------------------------- # Public API # ----------------------------------------------------------------------- async def get_recent_summary(self, n: int = 3) -> str: """ Fetches last N royal_conversations where kinan participated. Queries royal_conversations with: participants->>'kinan' = 'true' ORDER BY started_at DESC LIMIT %s Returns a human-readable string such as: RECENT CONTEXT: [2026-02-24] Call (18min): George's Cairns number checked. Action: Follow up. [2026-02-23] Call (6min): Voice widget confirmed live. Kinan happy. Returns empty string (never None) when: - No conversations exist - postgres_pool is None - Any database error occurs Args: n: Number of recent conversations to fetch (default 3, must be >= 1). Returns: Formatted string with RECENT CONTEXT header, or "" if nothing found. """ if self._pool is None: return "" if n < 1: return "" conn = None try: conn = self._pool.getconn() rows = self._fetch_rows(conn, n) except Exception: return "" finally: if conn is not None: try: self._pool.putconn(conn) except Exception: pass if not rows: return "" lines = [] for row in rows: line = _format_row(row) if line: lines.append(line) if not lines: return "" return "RECENT CONTEXT:\n" + "\n".join(lines) # ----------------------------------------------------------------------- # Internal helpers # ----------------------------------------------------------------------- def _fetch_rows(self, conn, n: int) -> list: """ Execute parameterised SELECT against royal_conversations. Uses LIMIT %s (not Python slice) so the database does the bounding. Returns a list of dict-like rows, each containing: started_at, ended_at, summary, action_items """ _SQL = ( "SELECT started_at, ended_at, summary, action_items " "FROM royal_conversations " "WHERE participants->>'kinan' = 'true' " "ORDER BY started_at DESC " "LIMIT %s" ) with conn.cursor() as cur: cur.execute(_SQL, (n,)) columns = [desc[0] for desc in cur.description] return [dict(zip(columns, row)) for row in cur.fetchall()] # --------------------------------------------------------------------------- # Row formatting helpers (module-level, pure functions, easy to unit test) # --------------------------------------------------------------------------- def _format_duration(started_at, ended_at) -> str: """ Return a human-readable duration string. If ended_at is None → "ongoing" Otherwise → "Xmin" where X is the integer number of minutes (rounded down). """ if ended_at is None: return "ongoing" try: delta = ended_at - started_at total_seconds = int(delta.total_seconds()) minutes = total_seconds // 60 return f"{minutes}min" except Exception: return "ongoing" def _format_action_items(action_items) -> str: """ Format action_items (JSONB array or None) into an inline string. Rules: - None / empty list / empty string → return "" (omit Action part) - JSON string that parses to a list → join with "; " - JSON string that parses to a single string → use as-is - Already a list (psycopg2 native) → join with "; " - Any other type → str() it Returns: " Action: " if there are items, else "". """ if action_items is None: return "" # psycopg2 may return a Python list directly when the column is JSONB items = action_items # If it's a raw JSON string, parse it if isinstance(action_items, str): stripped = action_items.strip() if not stripped or stripped in ("[]", "null", "{}"): return "" try: items = json.loads(stripped) except (json.JSONDecodeError, ValueError): # Not JSON — use the raw string as the action text return f" Action: {stripped}" # At this point items is either a list, dict, or other type if isinstance(items, list): # Filter out blanks non_empty = [str(i).strip() for i in items if str(i).strip()] if not non_empty: return "" return " Action: " + "; ".join(non_empty) if isinstance(items, dict): # Unexpected shape — convert to string text = str(items).strip() return f" Action: {text}" if text else "" text = str(items).strip() return f" Action: {text}" if text else "" def _format_row(row: dict) -> str: """ Format a single royal_conversations row into one summary line. Output format: [YYYY-MM-DD] Call (Xmin): Where action_part is either "" or " Action: ". Returns "" if started_at is missing. """ started_at = row.get("started_at") if started_at is None: return "" # Date portion try: date_str = started_at.strftime("%Y-%m-%d") except AttributeError: # Fallback if started_at is a string date_str = str(started_at)[:10] ended_at = row.get("ended_at") duration = _format_duration(started_at, ended_at) summary = (row.get("summary") or "").strip() if not summary: summary = "(no summary)" action_part = _format_action_items(row.get("action_items")) return f"[{date_str}] Call ({duration}): {summary}{action_part}" # VERIFICATION_STAMP # Story: 4.07 # Verified By: parallel-builder # Verified At: 2026-02-25T00:00:00Z # Tests: 9/9 # Coverage: 100%