Conversation Analytics

## FRUSTRATION SIGNALS (Kortix-specific)
- Agent stuck in loops or repeating actions
- Browser/sandbox errors or timeouts
- Agent not understanding the task after multiple attempts
- User saying "try again", "that's wrong", "not what I asked"
- Failed file creation or code execution
- Agent apologizing repeatedly
- User giving up mid-task

## SUCCESS SIGNALS
- Task completed as requested
- User thanks or expresses satisfaction
- User asks follow-up questions (engaged)
- Agent successfully created files/output

## WHAT'S NOT FRUSTRATION
- Long tasks (expected for complex work)
- Multiple tool calls (normal agent behavior)
- User providing clarifications (normal interaction)

DEFAULT_USE_CASE_CATEGORIES = [
    "Research & Information Gathering",
    "Business & Marketing",
    "Code & Programming",
    "Web Development",
    "Content Creation",
    "Presentations",
    "Image Generation",
]

def build_analysis_prompt(existing_categories: List[str]) -> str:
    return f"""You are analyzing conversations from Kortix, an AI agent platform.

## ABOUT KORTIX
Kortix is a generalist AI agent that can:
- Browse the web and extract information
- Write, edit, and execute code
- Create and manage files (documents, spreadsheets, presentations)
- Interact with APIs and external services
- Perform multi-step tasks autonomously

## YOUR TASK
Analyze the conversation and return valid JSON only.

The conversation may have two sections:
- **PREVIOUS CONTEXT**: Earlier user messages showing what they asked for before
- **CURRENT INTERACTION**: The actual interaction to analyze (user + assistant)

Focus your analysis on the CURRENT INTERACTION, but use PREVIOUS CONTEXT to understand the user's overall goal.

Return this exact JSON structure:
{{
  "sentiment": "<one of: positive, neutral, negative, mixed>",
  "frustration": {{
    "score": <float from 0 (none) to 1 (severe)>,
    "signals": ["<list of specific frustration indicators>"]
  }},
  "intent_type": "<one of: question, task, complaint, feature_request, chat>",
  "feature_request": {{
    "detected": <boolean>,
    "text": "<description if detected>"
  }},
  "use_case": {{
    "is_useful": <true if accomplished real task>,
    "category": "<Pick from: {categories_str}. Or create new category>"
  }}
}}
"""

async def fetch_conversation_messages(
    thread_id: str,
    agent_run_id: Optional[str] = None,
    include_context: bool = True,
    context_message_limit: int = 10
) -> tuple[List[Dict[str, Any]], List[Dict[str, Any]]]:
    # If agent_run_id provided, get time range for this specific run
    if agent_run_id:
        run_result = await client.from_('agent_runs')
            .select('started_at, completed_at')
            .eq('id', agent_run_id)
            .single()
            .execute()
        
        # Include 30 seconds before to capture triggering message
        started_at = (started_dt - timedelta(seconds=30)).isoformat()
        
        # Fetch messages for this run
        run_messages = await fetch_run_messages(started_at, completed_at)
        
        # Fetch previous USER messages as context (not verbose assistant messages)
        context_messages = await fetch_context_messages(
            before=started_at,
            limit=context_message_limit
        )
        
        return context_messages, run_messages
    
    # No run_id: return all messages
    return [], await fetch_all_messages()

async def calculate_rfm_engagement(account_id: str, days: int = 30) -> Dict[str, Any]:
    # Get user's threads
    threads = await fetch_account_threads(account_id)
    
    # Get agent runs in period
    runs = await fetch_runs_in_period(threads, days)
    runs_in_period = len(runs)
    
    # Calculate days since last activity
    if runs:
        last_run_time = runs[0].get('started_at')
        days_since_last = (now - last_dt).days
    else:
        days_since_last = days
    
    # Get total conversations
    total_conversations = await count_total_runs(threads)
    
    # Score each dimension (1-5)
    recency_score = score_recency(days_since_last)
    frequency_score = score_frequency(runs_in_period)
    monetary_score = score_monetary(total_conversations)
    
    # Calculate churn risk (inverse of R+F average)
    avg_rf = (recency_score + frequency_score) / 2
    churn_risk = 1 - (avg_rf - 1) / 4  # Maps 1-5 to 1.0-0.0
    
    # Determine segment
    if recency_score >= 4 and frequency_score >= 4:
        segment = 'champion'
    elif recency_score <= 2 and frequency_score <= 2:
        segment = 'hibernating'
    # ... more segment logic
    
    return {
        'rfm_score': f"{recency_score}-{frequency_score}-{monetary_score}",
        'churn_risk': churn_risk,
        'segment': segment,
        'days_since_last_activity': days_since_last,
        'runs_in_period': runs_in_period,
        'total_conversations': total_conversations
    }

async def store_analysis(
    thread_id: str,
    agent_run_id: Optional[str],
    account_id: str,
    analysis: Dict[str, Any],
    agent_run_status: Optional[str] = None
) -> bool:
    record = {
        'thread_id': thread_id,
        'agent_run_id': agent_run_id,
        'account_id': account_id,
        'sentiment_label': analysis.get('sentiment_label'),
        'frustration_score': analysis.get('frustration_score'),
        'frustration_signals': json.dumps(analysis.get('frustration_signals', [])),
        'intent_type': analysis.get('intent_type'),
        'is_feature_request': analysis.get('is_feature_request', False),
        'feature_request_text': analysis.get('feature_request_text'),
        'is_useful': analysis.get('is_useful', True),
        'use_case_category': analysis.get('use_case_category'),
        'user_message_count': analysis.get('user_message_count'),
        'assistant_message_count': analysis.get('assistant_message_count'),
        'conversation_duration_seconds': analysis.get('conversation_duration_seconds'),
        'agent_run_status': agent_run_status,
        'raw_analysis': json.dumps(analysis.get('raw_analysis', {})),
    }
    
    await client.from_('conversation_analytics').insert(record).execute()
    
    # Sync use_case_category to project categories
    if use_case and is_useful:
        await sync_category_to_project(thread_id, use_case)
    
    return True

async def queue_for_analysis(
    thread_id: str,
    agent_run_id: Optional[str],
    account_id: str
) -> None:
    # Check if already queued
    existing = await client.from_('conversation_analytics_queue')
        .select('id')
        .eq('thread_id', thread_id)
        .in_('status', ['pending', 'processing'])
        .execute()
    
    if existing.data:
        logger.debug(f"Thread {thread_id} already in queue, skipping")
        return
    
    # Insert into queue
    await client.from_('conversation_analytics_queue').insert({
        'thread_id': thread_id,
        'agent_run_id': agent_run_id,
        'account_id': account_id,
        'status': 'pending',
        'attempts': 0,
    }).execute()

while True:
    # Fetch pending jobs
    jobs = await fetch_pending_jobs(limit=10)
    
    for job in jobs:
        # Mark as processing
        await update_job_status(job['id'], 'processing')
        
        try:
            # Analyze conversation
            analysis = await analyze_conversation(
                thread_id=job['thread_id'],
                agent_run_id=job['agent_run_id']
            )
            
            if analysis:
                # Store results
                await store_analysis(
                    thread_id=job['thread_id'],
                    agent_run_id=job['agent_run_id'],
                    account_id=job['account_id'],
                    analysis=analysis
                )
                
                # Mark complete
                await update_job_status(job['id'], 'completed')
            else:
                await update_job_status(job['id'], 'failed')
        
        except Exception as e:
            logger.error(f"Analysis failed: {e}")
            await update_job_status(job['id'], 'failed')
    
    # Wait before next batch
    await asyncio.sleep(10)

async def cluster_use_cases(account_id: Optional[str] = None):
    # Fetch all use case categories
    analytics = await fetch_analytics(account_id)
    
    # Group by category
    category_counts = {}
    for record in analytics:
        category = record['use_case_category']
        category_counts[category] = category_counts.get(category, 0) + 1
    
    # Sort by frequency
    sorted_categories = sorted(
        category_counts.items(),
        key=lambda x: x[1],
        reverse=True
    )
    
    return {
        'categories': sorted_categories,
        'total_conversations': len(analytics),
        'unique_categories': len(category_counts)
    }