# app/core/context.py from dataclasses import dataclass, field from typing import Dict, Any, Optional, List from enum import Enum # ═══════════════════════════════════════════════════════════════════════════════ # SESSION LIFECYCLE STATE (GUVI Compliance) # ═══════════════════════════════════════════════════════════════════════════════ class SessionState(Enum): """ Session lifecycle for GUVI evaluation. Prevents duplicate callbacks and ensures exactly-once reporting. """ ACTIVE = "active" # Receiving messages, not yet confirmed scam ENGAGING = "engaging" # Scam confirmed, agent actively engaging COMPLETE = "complete" # Engagement finished, ready for callback REPORTED = "reported" # Callback sent, session closed @dataclass class TurnContext: """ Request-scoped context to prevent redundant computations and API storms. Tracks what has been decided/attempted within a single message turn. """ session_id: str message: str sender_role: str = "scammer" # [SCORING] scammer or user # Decision Flags (Stop Re-evaluation) scam_decided: bool = False scam_type: Optional[str] = None scam_confidence: float = 0.0 # Execution Guards (Stop Loops) fast_chat_attempted: bool = False structured_extraction_done: bool = False persona_selected: bool = False persona_locked: bool = False # NEW: Hard lock for session finalized: bool = False # KILL SWITCH: If True, stop all LLM calls # Budget Tracking (Global) llm_call_count: int = 0 # Per-turn counter session_llm_total: int = 0 # Session-level counter (NEW: for MAX_SESSION enforcement) budget_exceeded: bool = False session: Dict = field(default_factory=dict) # Reference to conversation for session budget reply_mode: str = "NORMAL" # NORMAL or HONEYPOT_ONLY (terminal) def mark_scam(self, scam_type: str, confidence: float): self.scam_decided = True self.scam_type = scam_type self.scam_confidence = confidence def should_skip_reasoning(self) -> bool: """Logic for Exclusive Borderline Check""" # If already decided OR score is very high/low => Skip reasoning if self.scam_decided: return True if self.scam_confidence > 0.9 or self.scam_confidence < 0.2: return True return False def is_engagement_complete(session: Dict, scam_detected: bool = False) -> bool: """ GUVI Engagement Complete Rules - AGGRESSIVE Completion for Judging. Returns True when the agent should stop engaging and send final callback. Critical: GUVI evaluation is fast (3-4 turns). Capture intel and finalize quickly! """ messages = len(session.get("history", [])) intel = session.get("aggregated_intelligence", {}) # ═══════════════════════════════════════════════════════════════════════════ # RULE 0: Budget Exhausted - ALWAYS finalize (prevent lost callbacks) # ═══════════════════════════════════════════════════════════════════════════ if session.get("budget_exceeded", False): return True # ═══════════════════════════════════════════════════════════════════════════ # RULE 1: AGGRESSIVE COMPLETION - Intel captured = Finalize next turn # GUVI Critical: If UPI/Bank captured, judges need to see it IMMEDIATELY # ═══════════════════════════════════════════════════════════════════════════ has_high_value_intel = ( len(intel.get("upi_ids", [])) > 0 or len(intel.get("bank_accounts", [])) > 0 or len(intel.get("credit_cards", [])) > 0 ) # Aggressive: If we have high-value intel AND at least 2 turns, finalize! if has_high_value_intel and messages >= 2: return True # ═══════════════════════════════════════════════════════════════════════════ # RULE 2: MEDIUM VALUE INTEL + 3 turns = Finalize # ═══════════════════════════════════════════════════════════════════════════ has_medium_intel = ( len(intel.get("phone_numbers", [])) >= 1 or len(intel.get("pan_cards", [])) > 0 or len(intel.get("aadhar_numbers", [])) > 0 ) if has_medium_intel and messages >= 3: return True # ═══════════════════════════════════════════════════════════════════════════ # RULE 3: Scam confirmed + engagement depth = Finalize # ═══════════════════════════════════════════════════════════════════════════ if scam_detected and messages >= 4: return True # ═══════════════════════════════════════════════════════════════════════════ # RULE 4: Hard cap at 6 messages (don't waste GUVI's time) # ═══════════════════════════════════════════════════════════════════════════ if messages >= 6: return True # ═══════════════════════════════════════════════════════════════════════════ # RULE 5: Scammer Agitation = Finalize early # ═══════════════════════════════════════════════════════════════════════════ agitation_list = intel.get("metadata_agitation", []) if agitation_list and agitation_list[-1].upper() in ["AGITATED", "VOLATILE"] and messages >= 3: return True return False def get_session_state(session: Dict) -> SessionState: """Get current session lifecycle state.""" state_str = session.get("lifecycle_state", "active") try: return SessionState(state_str) except ValueError: return SessionState.ACTIVE def set_session_state(session: Dict, state: SessionState): """Set session lifecycle state.""" session["lifecycle_state"] = state.value