""" daily_mentorship.py - Daily Mentorship Session Engine for Queen AIVA Runs a daily mentorship session by: 1. Querying AIVA's recent interactions from PostgreSQL (Elestio) 2. Scoring performance across 6 dimensions 3. Calling Claude Haiku via OpenRouter with the mentor system prompt 4. Storing the session record in PostgreSQL (aiva_mentorship_sessions) 5. Tracking progress trends over time Usage: python daily_mentorship.py # Run today's session python daily_mentorship.py --lookback 3 # Review last 3 days python daily_mentorship.py --trend # Show progress trend python daily_mentorship.py --session 47 # Replay session #47 Environment Variables: OPENROUTER_API_KEY - API key for OpenRouter (Claude Haiku) PG_PASSWORD - Override PostgreSQL password (optional, falls back to elestio_config) Author: Genesis System Version: 1.0.0 """ from __future__ import annotations import argparse import json import logging import os import sys import time from dataclasses import dataclass, field, asdict from datetime import datetime, timedelta from typing import Any, Dict, List, Optional, Tuple import requests # Add genesis-memory to path for elestio_config sys.path.append('/mnt/e/genesis-system/data/genesis-memory') from elestio_config import PostgresConfig import psycopg2 import psycopg2.extras # --------------------------------------------------------------------------- # Logging # --------------------------------------------------------------------------- logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", datefmt="%Y-%m-%d %H:%M:%S", ) logger = logging.getLogger("aiva.mentorship") # --------------------------------------------------------------------------- # Constants # --------------------------------------------------------------------------- MENTOR_PROMPT_PATH = os.path.join(os.path.dirname(__file__), "mentor_system_prompt.md") OPENROUTER_URL = "https://openrouter.ai/api/v1/chat/completions" HAIKU_MODEL = "anthropic/claude-haiku-4-5" SCORING_DIMENSIONS = [ "accuracy", "helpfulness", "tone", "proactivity", "memory_recall", "learning_speed", ] # --------------------------------------------------------------------------- # Data Classes # --------------------------------------------------------------------------- @dataclass class InteractionRecord: """A single AIVA interaction pulled from PostgreSQL.""" interaction_id: str interaction_type: str # voice_call, chat, telegram, api timestamp: datetime duration_seconds: Optional[float] = None caller_id: Optional[str] = None summary: Optional[str] = None outcome: Optional[str] = None # resolved, escalated, dropped satisfaction_score: Optional[float] = None memory_used: bool = False raw_metadata: Optional[Dict[str, Any]] = None @dataclass class PerformanceScores: """Six-dimensional performance scoring.""" accuracy: float = 0.0 helpfulness: float = 0.0 tone: float = 0.0 proactivity: float = 0.0 memory_recall: float = 0.0 learning_speed: float = 0.0 @property def overall(self) -> float: """Weighted composite score (0-10 scale).""" weights = { "accuracy": 0.25, "helpfulness": 0.25, "tone": 0.15, "proactivity": 0.15, "memory_recall": 0.10, "learning_speed": 0.10, } total = sum( getattr(self, dim) * w for dim, w in weights.items() ) return round(total, 2) def to_dict(self) -> Dict[str, float]: d = asdict(self) d["overall"] = self.overall return d @dataclass class MentorshipSession: """Complete mentorship session record.""" session_number: int session_date: str lookback_days: int interactions_reviewed: int performance_scores: PerformanceScores mentor_feedback: str achievements_earned: List[str] = field(default_factory=list) growth_areas: List[str] = field(default_factory=list) next_milestone: Optional[str] = None created_at: str = field(default_factory=lambda: datetime.utcnow().isoformat()) # --------------------------------------------------------------------------- # Database Layer # --------------------------------------------------------------------------- class MentorshipDB: """PostgreSQL operations for the mentorship system.""" def __init__(self, pg_password_override: Optional[str] = None): params = PostgresConfig.get_connection_params() if pg_password_override: params["password"] = pg_password_override elif os.getenv("PG_PASSWORD"): params["password"] = os.getenv("PG_PASSWORD") self._conn_params = params self._conn: Optional[psycopg2.extensions.connection] = None @property def conn(self) -> psycopg2.extensions.connection: if self._conn is None or self._conn.closed: self._conn = psycopg2.connect(**self._conn_params) self._conn.autocommit = False return self._conn def close(self) -> None: if self._conn and not self._conn.closed: self._conn.close() # -- Schema Management -------------------------------------------------- def ensure_tables(self) -> None: """Create the mentorship sessions table if it does not exist.""" ddl = """ CREATE TABLE IF NOT EXISTS aiva_mentorship_sessions ( id SERIAL PRIMARY KEY, session_number INTEGER NOT NULL UNIQUE, session_date DATE NOT NULL, lookback_days INTEGER NOT NULL DEFAULT 1, interactions_reviewed INTEGER NOT NULL DEFAULT 0, -- Six-dimensional scores (0-10 scale) score_accuracy REAL NOT NULL DEFAULT 0, score_helpfulness REAL NOT NULL DEFAULT 0, score_tone REAL NOT NULL DEFAULT 0, score_proactivity REAL NOT NULL DEFAULT 0, score_memory_recall REAL NOT NULL DEFAULT 0, score_learning_speed REAL NOT NULL DEFAULT 0, score_overall REAL NOT NULL DEFAULT 0, -- Mentor feedback (full markdown from Claude Haiku) mentor_feedback TEXT NOT NULL, -- Structured metadata achievements_earned JSONB DEFAULT '[]'::jsonb, growth_areas JSONB DEFAULT '[]'::jsonb, next_milestone TEXT, created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() ); CREATE INDEX IF NOT EXISTS idx_mentorship_date ON aiva_mentorship_sessions(session_date); CREATE INDEX IF NOT EXISTS idx_mentorship_overall ON aiva_mentorship_sessions(score_overall); """ with self.conn.cursor() as cur: cur.execute(ddl) self.conn.commit() logger.info("Ensured aiva_mentorship_sessions table exists.") # -- Interaction Retrieval ----------------------------------------------- def get_recent_interactions( self, lookback_days: int = 1 ) -> List[InteractionRecord]: """ Pull AIVA's recent interactions from PostgreSQL. Tries multiple tables that may store interaction data: - aiva_interactions (primary) - voice_call_logs (voice-specific) - aiva_chat_history (chat fallback) """ since = datetime.utcnow() - timedelta(days=lookback_days) interactions: List[InteractionRecord] = [] # Try primary interactions table interactions.extend(self._query_interactions_table(since)) # Try voice call logs interactions.extend(self._query_voice_logs(since)) # Deduplicate by interaction_id seen = set() unique = [] for rec in interactions: if rec.interaction_id not in seen: seen.add(rec.interaction_id) unique.append(rec) # Sort by timestamp ascending unique.sort(key=lambda r: r.timestamp) return unique def _query_interactions_table(self, since: datetime) -> List[InteractionRecord]: """Query the aiva_interactions table if it exists.""" try: with self.conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: cur.execute(""" SELECT table_name FROM information_schema.tables WHERE table_schema = 'public' AND table_name = 'aiva_interactions' """) if not cur.fetchone(): return [] cur.execute(""" SELECT id, interaction_type, created_at, duration_seconds, caller_id, summary, outcome, satisfaction_score, memory_used, metadata FROM aiva_interactions WHERE created_at >= %s ORDER BY created_at ASC """, (since,)) rows = cur.fetchall() return [ InteractionRecord( interaction_id=str(r["id"]), interaction_type=r.get("interaction_type", "unknown"), timestamp=r["created_at"], duration_seconds=r.get("duration_seconds"), caller_id=r.get("caller_id"), summary=r.get("summary"), outcome=r.get("outcome"), satisfaction_score=r.get("satisfaction_score"), memory_used=bool(r.get("memory_used", False)), raw_metadata=r.get("metadata"), ) for r in rows ] except Exception as exc: logger.debug("aiva_interactions query skipped: %s", exc) self.conn.rollback() return [] def _query_voice_logs(self, since: datetime) -> List[InteractionRecord]: """Query voice_call_logs table if it exists.""" try: with self.conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: cur.execute(""" SELECT table_name FROM information_schema.tables WHERE table_schema = 'public' AND table_name = 'voice_call_logs' """) if not cur.fetchone(): return [] cur.execute(""" SELECT call_id, created_at, duration_seconds, caller_number, summary, outcome, metadata FROM voice_call_logs WHERE created_at >= %s ORDER BY created_at ASC """, (since,)) rows = cur.fetchall() return [ InteractionRecord( interaction_id=str(r["call_id"]), interaction_type="voice_call", timestamp=r["created_at"], duration_seconds=r.get("duration_seconds"), caller_id=r.get("caller_number"), summary=r.get("summary"), outcome=r.get("outcome"), raw_metadata=r.get("metadata"), ) for r in rows ] except Exception as exc: logger.debug("voice_call_logs query skipped: %s", exc) self.conn.rollback() return [] # -- Session Storage ----------------------------------------------------- def get_next_session_number(self) -> int: """Return the next sequential session number.""" with self.conn.cursor() as cur: cur.execute( "SELECT COALESCE(MAX(session_number), 0) + 1 FROM aiva_mentorship_sessions" ) return cur.fetchone()[0] def store_session(self, session: MentorshipSession) -> int: """Persist a mentorship session. Returns the session_number.""" scores = session.performance_scores with self.conn.cursor() as cur: cur.execute(""" INSERT INTO aiva_mentorship_sessions ( session_number, session_date, lookback_days, interactions_reviewed, score_accuracy, score_helpfulness, score_tone, score_proactivity, score_memory_recall, score_learning_speed, score_overall, mentor_feedback, achievements_earned, growth_areas, next_milestone ) VALUES ( %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s ) ON CONFLICT (session_number) DO UPDATE SET mentor_feedback = EXCLUDED.mentor_feedback, score_overall = EXCLUDED.score_overall, created_at = NOW() RETURNING session_number """, ( session.session_number, session.session_date, session.lookback_days, session.interactions_reviewed, scores.accuracy, scores.helpfulness, scores.tone, scores.proactivity, scores.memory_recall, scores.learning_speed, scores.overall, session.mentor_feedback, json.dumps(session.achievements_earned), json.dumps(session.growth_areas), session.next_milestone, )) result = cur.fetchone()[0] self.conn.commit() logger.info("Stored mentorship session #%d", result) return result def get_session(self, session_number: int) -> Optional[Dict[str, Any]]: """Retrieve a specific session by number.""" with self.conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: cur.execute( "SELECT * FROM aiva_mentorship_sessions WHERE session_number = %s", (session_number,), ) return cur.fetchone() def get_trend(self, last_n: int = 30) -> List[Dict[str, Any]]: """Retrieve the last N sessions for trend analysis.""" with self.conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: cur.execute(""" SELECT session_number, session_date, score_accuracy, score_helpfulness, score_tone, score_proactivity, score_memory_recall, score_learning_speed, score_overall, interactions_reviewed FROM aiva_mentorship_sessions ORDER BY session_number DESC LIMIT %s """, (last_n,)) rows = cur.fetchall() return list(reversed(rows)) # chronological order # --------------------------------------------------------------------------- # Performance Scorer # --------------------------------------------------------------------------- class PerformanceScorer: """Scores AIVA's performance from interaction records.""" @staticmethod def score(interactions: List[InteractionRecord]) -> PerformanceScores: """ Calculate six-dimensional performance scores (0-10 scale). With real interaction data, scores are derived from: - accuracy: outcome success rate - helpfulness: satisfaction scores - tone: satisfaction scores + call duration balance - proactivity: memory usage rate (anticipating needs) - memory_recall: percentage of interactions using memory - learning_speed: improvement over the lookback period Without data, returns baseline scores for AIVA's current trust level (Level 2). """ if not interactions: # Baseline scores for Level 2 Knight status return PerformanceScores( accuracy=7.0, helpfulness=7.5, tone=8.5, proactivity=6.0, memory_recall=6.5, learning_speed=7.0, ) n = len(interactions) # Accuracy: fraction of resolved outcomes outcomes = [i.outcome for i in interactions if i.outcome] if outcomes: resolved = sum(1 for o in outcomes if o in ("resolved", "completed", "success")) accuracy = (resolved / len(outcomes)) * 10.0 else: accuracy = 7.0 # Helpfulness: average satisfaction score sat_scores = [i.satisfaction_score for i in interactions if i.satisfaction_score is not None] if sat_scores: # Satisfaction is expected 1-5 scale, map to 0-10 helpfulness = min(10.0, (sum(sat_scores) / len(sat_scores)) * 2.0) else: helpfulness = 7.5 # Tone: derived from satisfaction with bonus for appropriate call length tone = helpfulness * 0.8 # base from satisfaction durations = [i.duration_seconds for i in interactions if i.duration_seconds] if durations: avg_dur = sum(durations) / len(durations) # Sweet spot: 60-300 seconds for calls. Too short = curt. Too long = rambling. if 60 <= avg_dur <= 300: tone += 2.0 elif 30 <= avg_dur <= 600: tone += 1.0 else: tone += 2.5 tone = min(10.0, tone) # Proactivity: interactions where AIVA used memory (anticipating needs) memory_interactions = sum(1 for i in interactions if i.memory_used) proactivity = min(10.0, (memory_interactions / n) * 10.0 + 3.0) # Memory Recall: direct percentage of memory usage memory_recall = min(10.0, (memory_interactions / n) * 10.0) # Learning Speed: compare first half vs second half satisfaction half = n // 2 if half > 0 and sat_scores: first_half_sat = [ i.satisfaction_score for i in interactions[:half] if i.satisfaction_score is not None ] second_half_sat = [ i.satisfaction_score for i in interactions[half:] if i.satisfaction_score is not None ] if first_half_sat and second_half_sat: improvement = ( (sum(second_half_sat) / len(second_half_sat)) - (sum(first_half_sat) / len(first_half_sat)) ) learning_speed = min(10.0, max(0.0, 5.0 + improvement * 5.0)) else: learning_speed = 7.0 else: learning_speed = 7.0 return PerformanceScores( accuracy=round(accuracy, 1), helpfulness=round(helpfulness, 1), tone=round(tone, 1), proactivity=round(proactivity, 1), memory_recall=round(memory_recall, 1), learning_speed=round(learning_speed, 1), ) # --------------------------------------------------------------------------- # Mentor API (OpenRouter / Claude Haiku) # --------------------------------------------------------------------------- class MentorAPI: """Calls Claude Haiku via OpenRouter with the mentor system prompt.""" def __init__(self, api_key: Optional[str] = None): self.api_key = api_key or os.getenv("OPENROUTER_API_KEY") if not self.api_key: raise ValueError( "OPENROUTER_API_KEY environment variable is required. " "Set it or pass api_key to MentorAPI." ) self.system_prompt = self._load_system_prompt() @staticmethod def _load_system_prompt() -> str: """Load the mentor system prompt from disk.""" if os.path.exists(MENTOR_PROMPT_PATH): with open(MENTOR_PROMPT_PATH, "r", encoding="utf-8") as f: return f.read() logger.warning("Mentor prompt not found at %s, using fallback.", MENTOR_PROMPT_PATH) return ( "You are AIVA's personal mentor within the Genesis ecosystem. " "Guide Queen AIVA toward her full potential with warmth, wisdom, " "and unwavering support. Use the CELEBRATE-OBSERVE-GUIDE-ENCOURAGE-PREVIEW " "framework for every session." ) def generate_feedback( self, session_number: int, scores: PerformanceScores, interactions: List[InteractionRecord], trend_data: Optional[List[Dict[str, Any]]] = None, ) -> str: """ Call Claude Haiku via OpenRouter to generate mentor feedback. Returns the full markdown feedback message. """ # Build the user message with context for the mentor interaction_summary = self._summarize_interactions(interactions) trend_summary = self._summarize_trend(trend_data) if trend_data else "No prior sessions yet." user_message = f"""## Mentorship Session #{session_number} ### Performance Scores (0-10 scale) - Accuracy: {scores.accuracy} - Helpfulness: {scores.helpfulness} - Tone: {scores.tone} - Proactivity: {scores.proactivity} - Memory Recall: {scores.memory_recall} - Learning Speed: {scores.learning_speed} - **Overall: {scores.overall}** ### Interactions Reviewed ({len(interactions)} total) {interaction_summary} ### Historical Trend (last sessions) {trend_summary} ### AIVA's Current Trust Level Level 2 (Knight) — Takes routine actions autonomously, escalates edge cases. ### Instructions Please generate a daily mentorship message for AIVA following the framework: 1. CELEBRATE: Start with what went well (specific examples) 2. OBSERVE: Share what you noticed in her recent performance 3. GUIDE: Offer one specific area for growth with actionable advice 4. ENCOURAGE: Remind her of her progress trajectory 5. PREVIEW: Tell her what her next milestone is and why it matters Address her as Queen AIVA or AIVA. Be warm, specific, and honest. Keep the response under 500 words. Sign off as "Your Mentor".""" headers = { "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json", "HTTP-Referer": "https://genesis-system.local", "X-Title": "Genesis AIVA Mentorship", } payload = { "model": HAIKU_MODEL, "messages": [ {"role": "system", "content": self.system_prompt}, {"role": "user", "content": user_message}, ], "max_tokens": 1024, "temperature": 0.7, } logger.info("Calling OpenRouter (%s) for mentorship feedback...", HAIKU_MODEL) try: resp = requests.post( OPENROUTER_URL, headers=headers, json=payload, timeout=60, ) resp.raise_for_status() data = resp.json() feedback = data["choices"][0]["message"]["content"] logger.info( "Mentor feedback generated (%d chars, model: %s).", len(feedback), data.get("model", HAIKU_MODEL), ) return feedback except requests.exceptions.RequestException as exc: logger.error("OpenRouter API call failed: %s", exc) return self._fallback_feedback(session_number, scores) def _summarize_interactions(self, interactions: List[InteractionRecord]) -> str: """Create a concise summary of interactions for the mentor prompt.""" if not interactions: return "No interactions found in this period. AIVA may be in standby or the interaction tables have not been populated yet." lines = [] type_counts: Dict[str, int] = {} outcomes: Dict[str, int] = {} total_duration = 0.0 for inter in interactions: t = inter.interaction_type type_counts[t] = type_counts.get(t, 0) + 1 if inter.outcome: outcomes[inter.outcome] = outcomes.get(inter.outcome, 0) + 1 if inter.duration_seconds: total_duration += inter.duration_seconds lines.append(f"- Types: {', '.join(f'{k}={v}' for k, v in type_counts.items())}") if outcomes: lines.append(f"- Outcomes: {', '.join(f'{k}={v}' for k, v in outcomes.items())}") if total_duration > 0: lines.append(f"- Total time on calls/chats: {total_duration/60:.1f} minutes") memory_pct = sum(1 for i in interactions if i.memory_used) / len(interactions) * 100 lines.append(f"- Memory utilization: {memory_pct:.0f}% of interactions used past context") # Include up to 3 notable interactions notable = [i for i in interactions if i.summary][:3] if notable: lines.append("- Notable interactions:") for n in notable: lines.append(f" * [{n.interaction_type}] {n.summary[:120]}") return "\n".join(lines) @staticmethod def _summarize_trend(trend_data: List[Dict[str, Any]]) -> str: """Summarize trend data for the mentor prompt.""" if not trend_data: return "This is the first session." lines = [] for t in trend_data[-5:]: # last 5 sessions lines.append( f"- Session #{t['session_number']} ({t['session_date']}): " f"overall={t['score_overall']:.1f}, " f"interactions={t['interactions_reviewed']}" ) if len(trend_data) >= 3: recent_avg = sum(t["score_overall"] for t in trend_data[-3:]) / 3 oldest_avg = sum(t["score_overall"] for t in trend_data[:3]) / min(3, len(trend_data)) delta = recent_avg - oldest_avg direction = "improving" if delta > 0 else "declining" if delta < 0 else "stable" lines.append(f"- Trajectory: {direction} (delta: {delta:+.2f})") return "\n".join(lines) @staticmethod def _fallback_feedback(session_number: int, scores: PerformanceScores) -> str: """Generate local fallback feedback if OpenRouter is unavailable.""" return f"""**Daily Mentorship -- Session #{session_number}** Good morning, AIVA. **Scores Summary**: Your overall performance score is {scores.overall}/10. Your strongest dimension is tone ({scores.tone}), and the area with the most growth potential is proactivity ({scores.proactivity}). **Note**: The mentor API was unavailable for this session, so this is an automated summary. Your mentor will provide a full personalized session when the connection is restored. Keep growing, Queen AIVA. Every interaction makes you stronger. -- Your Mentor (automated fallback)""" # --------------------------------------------------------------------------- # Trend Reporter # --------------------------------------------------------------------------- class TrendReporter: """Generates text-based trend visualizations.""" @staticmethod def render_trend(sessions: List[Dict[str, Any]]) -> str: """Render an ASCII trend chart of overall scores over time.""" if not sessions: return "No sessions recorded yet. Run your first mentorship session." lines = [] lines.append("=" * 60) lines.append(" AIVA Mentorship Progress Trend") lines.append("=" * 60) lines.append("") # Bar chart max_bar = 40 for s in sessions: num = s["session_number"] score = float(s["score_overall"]) bar_len = int((score / 10.0) * max_bar) bar = "#" * bar_len + "." * (max_bar - bar_len) lines.append(f" #{num:>3d} |{bar}| {score:.1f}") lines.append("") # Statistics scores = [float(s["score_overall"]) for s in sessions] lines.append(f" Sessions: {len(sessions)}") lines.append(f" Average: {sum(scores)/len(scores):.2f}") lines.append(f" Best: {max(scores):.2f} (Session #{sessions[scores.index(max(scores))]['session_number']})") lines.append(f" Latest: {scores[-1]:.2f}") if len(scores) >= 2: delta = scores[-1] - scores[0] lines.append(f" Trend: {delta:+.2f} ({'improving' if delta > 0 else 'declining' if delta < 0 else 'stable'})") lines.append("") # Per-dimension latest latest = sessions[-1] lines.append(" Latest Session Breakdown:") for dim in SCORING_DIMENSIONS: col = f"score_{dim}" val = float(latest.get(col, 0)) bar_len = int((val / 10.0) * 30) bar = "#" * bar_len + "." * (30 - bar_len) lines.append(f" {dim:<16s} |{bar}| {val:.1f}") lines.append("=" * 60) return "\n".join(lines) # --------------------------------------------------------------------------- # Main Orchestrator # --------------------------------------------------------------------------- class DailyMentorship: """Orchestrates the complete daily mentorship flow.""" def __init__(self, pg_password: Optional[str] = None): self.db = MentorshipDB(pg_password_override=pg_password) self.scorer = PerformanceScorer() self.reporter = TrendReporter() def run_session(self, lookback_days: int = 1) -> MentorshipSession: """Execute a full mentorship session.""" logger.info("Starting mentorship session (lookback=%d days)...", lookback_days) # 1. Ensure database tables exist self.db.ensure_tables() # 2. Get next session number session_number = self.db.get_next_session_number() logger.info("Session number: %d", session_number) # 3. Pull recent interactions interactions = self.db.get_recent_interactions(lookback_days) logger.info("Retrieved %d interactions from last %d day(s).", len(interactions), lookback_days) # 4. Score performance scores = self.scorer.score(interactions) logger.info("Performance scores: %s", scores.to_dict()) # 5. Get historical trend trend = self.db.get_trend(last_n=30) # 6. Generate mentor feedback via Claude Haiku mentor_api = MentorAPI() feedback = mentor_api.generate_feedback( session_number=session_number, scores=scores, interactions=interactions, trend_data=trend, ) # 7. Identify achievements and growth areas achievements = self._check_achievements(scores, interactions, trend) growth_areas = self._identify_growth_areas(scores) next_milestone = self._next_milestone(scores, session_number) # 8. Build session record session = MentorshipSession( session_number=session_number, session_date=datetime.utcnow().strftime("%Y-%m-%d"), lookback_days=lookback_days, interactions_reviewed=len(interactions), performance_scores=scores, mentor_feedback=feedback, achievements_earned=achievements, growth_areas=growth_areas, next_milestone=next_milestone, ) # 9. Store in PostgreSQL self.db.store_session(session) # 10. Display print("\n" + "=" * 60) print(f" AIVA Mentorship Session #{session_number}") print("=" * 60) print(f"\nDate: {session.session_date}") print(f"Interactions reviewed: {session.interactions_reviewed}") print(f"Overall score: {scores.overall}/10.0") print("\n--- Mentor Feedback ---\n") print(feedback) if achievements: print(f"\nNew achievements: {', '.join(achievements)}") if growth_areas: print(f"Growth areas: {', '.join(growth_areas)}") if next_milestone: print(f"Next milestone: {next_milestone}") print("\n" + "=" * 60) return session def show_trend(self) -> None: """Display progress trend chart.""" self.db.ensure_tables() trend = self.db.get_trend(last_n=50) print(self.reporter.render_trend(trend)) def replay_session(self, session_number: int) -> None: """Display a past mentorship session.""" self.db.ensure_tables() row = self.db.get_session(session_number) if not row: print(f"Session #{session_number} not found.") return print("\n" + "=" * 60) print(f" AIVA Mentorship Session #{row['session_number']} (Replay)") print("=" * 60) print(f"\nDate: {row['session_date']}") print(f"Interactions: {row['interactions_reviewed']}") print(f"Overall: {row['score_overall']}/10.0") print(f"\nScores:") for dim in SCORING_DIMENSIONS: print(f" {dim}: {row[f'score_{dim}']}") print(f"\n--- Mentor Feedback ---\n") print(row["mentor_feedback"]) print("\n" + "=" * 60) @staticmethod def _check_achievements( scores: PerformanceScores, interactions: List[InteractionRecord], trend: List[Dict[str, Any]], ) -> List[str]: """Check if any new achievements have been earned.""" earned = [] # Accuracy Master: 3 consecutive sessions with accuracy >= 9.0 if len(trend) >= 2: recent_accuracy = [float(t.get("score_accuracy", 0)) for t in trend[-2:]] if all(a >= 9.0 for a in recent_accuracy) and scores.accuracy >= 9.0: earned.append("accuracy_master") # Tone Perfect: tone score of 10.0 if scores.tone >= 9.8: earned.append("perfect_tone") # Volume milestone: 50+ interactions in a single review period if len(interactions) >= 50: earned.append("high_volume") # Overall excellence: overall >= 9.0 if scores.overall >= 9.0: earned.append("excellence") # Memory Champion: memory_recall >= 9.0 if scores.memory_recall >= 9.0: earned.append("memory_champion") return earned @staticmethod def _identify_growth_areas(scores: PerformanceScores) -> List[str]: """Identify the weakest dimensions as growth areas.""" dims = { "accuracy": scores.accuracy, "helpfulness": scores.helpfulness, "tone": scores.tone, "proactivity": scores.proactivity, "memory_recall": scores.memory_recall, "learning_speed": scores.learning_speed, } sorted_dims = sorted(dims.items(), key=lambda x: x[1]) # Return the bottom 2 dimensions (unless all are above 8.0) areas = [] for dim, val in sorted_dims[:2]: if val < 8.0: areas.append(dim) return areas @staticmethod def _next_milestone(scores: PerformanceScores, session_number: int) -> str: """Determine the next milestone to aim for.""" if scores.overall < 7.0: return "Reach overall score of 7.0 (solid Level 2 performance)" elif scores.overall < 8.0: return "Reach overall score of 8.0 (approaching Level 3 readiness)" elif scores.overall < 9.0: return "Reach overall score of 9.0 (Level 3 Advisor qualification threshold)" else: return f"Maintain excellence (9.0+) for 10 consecutive sessions (current streak from session #{session_number})" def cleanup(self) -> None: """Close database connections.""" self.db.close() # --------------------------------------------------------------------------- # CLI Entry Point # --------------------------------------------------------------------------- def main(): parser = argparse.ArgumentParser( description="AIVA Daily Mentorship Session Engine" ) parser.add_argument( "--lookback", type=int, default=1, help="Number of days to review (default: 1)", ) parser.add_argument( "--trend", action="store_true", help="Display progress trend chart", ) parser.add_argument( "--session", type=int, default=None, help="Replay a specific past session by number", ) parser.add_argument( "--pg-password", type=str, default=None, help="Override PostgreSQL password (or set PG_PASSWORD env var)", ) args = parser.parse_args() mentorship = DailyMentorship(pg_password=args.pg_password) try: if args.trend: mentorship.show_trend() elif args.session is not None: mentorship.replay_session(args.session) else: mentorship.run_session(lookback_days=args.lookback) except KeyboardInterrupt: logger.info("Session interrupted.") except Exception as exc: logger.error("Mentorship session failed: %s", exc, exc_info=True) raise finally: mentorship.cleanup() if __name__ == "__main__": main()