#!/usr/bin/env python3 """ Work Context Intelligence - Tracks current projects, pending threads, and priorities. This module maintains awareness of the steward's current work context, including: - Active projects and their status - Pending tasks and threads - Priority analysis and recommendations - Work continuity across sessions """ import json import os from datetime import datetime, timedelta from typing import Dict, List, Any, Optional, Tuple from collections import defaultdict, Counter from pathlib import Path import re import sys sys.path.insert(0, str(Path(__file__).parent.parent)) from config import MEMORY_DIR from utils.logging import setup_logger from conversations.comprehensive_indexer import ComprehensiveIndexer from core.memory.memory_manager import MemoryManager from utils.timeout_utils import with_timeout, safe_call_with_timeout, database_timeout, file_timeout logger = setup_logger(__name__) class WorkContextIntelligence: """Analyzes and tracks work context across sessions.""" def __init__(self): self.context_dir = os.path.join(MEMORY_DIR, "work_context") os.makedirs(self.context_dir, exist_ok=True) # Initialize components self.memory_manager = MemoryManager() self.conversation_indexer = ComprehensiveIndexer() # Load persistent context self.work_context = self._load_work_context() def _load_work_context(self) -> Dict[str, Any]: """Load persistent work context from disk.""" context_file = os.path.join(self.context_dir, "current_context.json") if os.path.exists(context_file): try: with open(context_file, 'r') as f: return json.load(f) except Exception as e: logger.error(f"Error loading work context: {e}") # Default context structure return { "active_projects": {}, "pending_threads": [], "recent_topics": [], "priority_stack": [], "work_sessions": [], "momentum_indicators": {}, "last_updated": datetime.now().isoformat() } def _save_work_context(self): """Save work context to disk.""" try: context_file = os.path.join(self.context_dir, "current_context.json") self.work_context["last_updated"] = datetime.now().isoformat() with open(context_file, 'w') as f: json.dump(self.work_context, f, indent=2) except Exception as e: logger.error(f"Error saving work context: {e}") @with_timeout(30, default_return={}, raise_on_timeout=False) # 30 second timeout def analyze_current_context(self, hours_back: int = 48) -> Dict[str, Any]: """ Analyze current work context from recent conversations and memories. Returns comprehensive context analysis. """ # Get recent conversations recent_messages = self._get_recent_conversations(hours_back) # Extract projects and topics projects = self._extract_projects(recent_messages) topics = self._extract_topics(recent_messages) threads = self._identify_pending_threads(recent_messages) # Analyze work patterns work_patterns = self._analyze_work_patterns(recent_messages) # Update context self._update_active_projects(projects) self._update_pending_threads(threads) self._update_recent_topics(topics) # Determine priorities priorities = self._analyze_priorities() # Build comprehensive analysis analysis = { "timestamp": datetime.now().isoformat(), "active_projects": self.work_context["active_projects"], "pending_threads": self.work_context["pending_threads"][:10], # Top 10 "recent_topics": self.work_context["recent_topics"][:20], # Top 20 "priorities": priorities, "work_patterns": work_patterns, "momentum_indicators": self._calculate_momentum_indicators(), "recommendations": self._generate_recommendations() } self._save_work_context() return analysis @database_timeout(default_return=[], raise_on_timeout=False) def _get_recent_conversations(self, hours_back: int) -> List[Dict]: """Get recent conversation messages.""" try: # Use conversation indexer to get recent messages cutoff_time = datetime.now() - timedelta(hours=hours_back) all_messages = self.conversation_indexer.quick_recall("", limit=1000) recent_messages = [] for msg in all_messages: try: # Parse timestamp from message if isinstance(msg, dict) and "timestamp" in msg: msg_time = datetime.fromisoformat(msg["timestamp"]) if msg_time >= cutoff_time: recent_messages.append(msg) except: continue return recent_messages except Exception as e: logger.error(f"Error getting recent conversations: {e}") return [] def _extract_projects(self, messages: List[Dict]) -> List[Dict[str, Any]]: """Extract project references from messages.""" projects = defaultdict(lambda: {"count": 0, "last_mentioned": None, "keywords": []}) # Common project indicators project_patterns = [ r"project[:\s]+([A-Za-z0-9\-_]+)", r"working on\s+([A-Za-z0-9\-_]+)", r"([A-Z][A-Za-z0-9\-_]+)\s+project", r"repository[:\s]+([A-Za-z0-9\-_]+)", r"codebase[:\s]+([A-Za-z0-9\-_]+)" ] for msg in messages: content = msg.get("content", "") timestamp = msg.get("timestamp", datetime.now().isoformat()) # Look for explicit project references for pattern in project_patterns: matches = re.findall(pattern, content, re.IGNORECASE) for match in matches: project_name = match.strip() projects[project_name]["count"] += 1 projects[project_name]["last_mentioned"] = timestamp # Extract keywords associated with projects if "MIRA" in content: projects["MIRA"]["count"] += 1 projects["MIRA"]["last_mentioned"] = timestamp # Extract MIRA-related keywords mira_keywords = re.findall(r"MIRA\s+(\w+)", content) projects["MIRA"]["keywords"].extend(mira_keywords) # Convert to list format project_list = [] for name, data in projects.items(): project_list.append({ "name": name, "activity_count": data["count"], "last_activity": data["last_mentioned"], "keywords": list(set(data["keywords"]))[:10] # Top 10 unique keywords }) # Sort by activity project_list.sort(key=lambda x: x["activity_count"], reverse=True) return project_list def _extract_topics(self, messages: List[Dict]) -> List[str]: """Extract discussion topics from messages.""" topics = Counter() # Topic extraction patterns topic_patterns = [ r"about\s+(\w+)", r"discussing\s+(\w+)", r"implement(?:ing)?\s+(\w+)", r"working on\s+(\w+)", r"(?:need|want) to\s+(\w+)" ] # Common technical topics to track tech_topics = [ "testing", "documentation", "refactoring", "optimization", "debugging", "feature", "bug", "performance", "security", "architecture", "design", "pattern", "integration", "deployment" ] for msg in messages: content = msg.get("content", "").lower() # Extract topics using patterns for pattern in topic_patterns: matches = re.findall(pattern, content) topics.update(matches) # Check for technical topics for topic in tech_topics: if topic in content: topics[topic] += 1 # Return sorted list of topics return [topic for topic, _ in topics.most_common(50)] def _identify_pending_threads(self, messages: List[Dict]) -> List[Dict[str, Any]]: """Identify unresolved threads and pending items.""" threads = [] # Patterns indicating pending work pending_patterns = [ (r"TODO[:\s]+(.+)", "todo"), (r"FIXME[:\s]+(.+)", "fixme"), (r"need to\s+(.+)", "need"), (r"should\s+(.+)", "should"), (r"will\s+(.+)", "future"), (r"plan to\s+(.+)", "planned"), (r"next[:\s]+(.+)", "next"), (r"pending[:\s]+(.+)", "pending") ] # Question patterns that might need follow-up question_patterns = [ r"how (?:do|can|should) (?:I|we)\s+(.+)\?", r"what (?:is|are|about)\s+(.+)\?", r"why (?:is|does|did)\s+(.+)\?" ] seen_threads = set() for msg in messages: content = msg.get("content", "") timestamp = msg.get("timestamp", datetime.now().isoformat()) role = msg.get("role", "user") # Extract pending items for pattern, thread_type in pending_patterns: matches = re.findall(pattern, content, re.IGNORECASE) for match in matches: thread_text = match.strip()[:200] # Limit length # Avoid duplicates thread_key = f"{thread_type}:{thread_text[:50]}" if thread_key not in seen_threads: seen_threads.add(thread_key) threads.append({ "type": thread_type, "content": thread_text, "timestamp": timestamp, "source_role": role, "priority": self._assess_thread_priority(thread_text, thread_type) }) # Extract unanswered questions if role == "user": for pattern in question_patterns: matches = re.findall(pattern, content, re.IGNORECASE) for match in matches: question = match.strip()[:200] thread_key = f"question:{question[:50]}" if thread_key not in seen_threads: seen_threads.add(thread_key) threads.append({ "type": "question", "content": f"Question: {question}", "timestamp": timestamp, "source_role": role, "priority": "medium" }) # Sort by priority and recency priority_order = {"critical": 0, "high": 1, "medium": 2, "low": 3} threads.sort(key=lambda x: ( priority_order.get(x["priority"], 3), x["timestamp"] ), reverse=True) return threads[:50] # Top 50 threads def _assess_thread_priority(self, content: str, thread_type: str) -> str: """Assess priority of a pending thread.""" content_lower = content.lower() # Critical indicators if any(word in content_lower for word in ["critical", "urgent", "asap", "immediately"]): return "critical" # High priority indicators if thread_type in ["fixme", "todo"] or any(word in content_lower for word in ["bug", "error", "broken", "failing"]): return "high" # Low priority indicators if any(word in content_lower for word in ["eventually", "someday", "maybe", "consider"]): return "low" return "medium" def _analyze_work_patterns(self, messages: List[Dict]) -> Dict[str, Any]: """Analyze patterns in work activity.""" patterns = { "session_times": [], "message_frequency": defaultdict(int), "work_types": defaultdict(int), "collaboration_level": 0 } # Analyze message timing for msg in messages: try: timestamp = datetime.fromisoformat(msg.get("timestamp", "")) hour = timestamp.hour patterns["message_frequency"][hour] += 1 # Categorize work type content = msg.get("content", "").lower() if any(word in content for word in ["implement", "code", "function", "class"]): patterns["work_types"]["coding"] += 1 elif any(word in content for word in ["debug", "error", "fix", "issue"]): patterns["work_types"]["debugging"] += 1 elif any(word in content for word in ["design", "architecture", "plan"]): patterns["work_types"]["designing"] += 1 elif any(word in content for word in ["test", "verify", "check"]): patterns["work_types"]["testing"] += 1 elif any(word in content for word in ["document", "readme", "comment"]): patterns["work_types"]["documenting"] += 1 except: continue # Calculate collaboration level user_messages = sum(1 for msg in messages if msg.get("role") == "user") assistant_messages = sum(1 for msg in messages if msg.get("role") == "assistant") if user_messages > 0: patterns["collaboration_level"] = min(assistant_messages / user_messages, 2.0) return patterns def _update_active_projects(self, projects: List[Dict]): """Update active projects tracking.""" current_time = datetime.now().isoformat() for project in projects[:10]: # Track top 10 projects name = project["name"] if name not in self.work_context["active_projects"]: self.work_context["active_projects"][name] = { "first_seen": current_time, "status": "active", "total_activity": 0 } # Update project info proj_data = self.work_context["active_projects"][name] proj_data["last_activity"] = project["last_activity"] proj_data["total_activity"] += project["activity_count"] proj_data["recent_keywords"] = project["keywords"] # Update status based on activity if project["activity_count"] > 5: proj_data["status"] = "highly_active" elif project["activity_count"] > 0: proj_data["status"] = "active" # Mark inactive projects for name, data in self.work_context["active_projects"].items(): if name not in [p["name"] for p in projects]: try: last_activity = datetime.fromisoformat(data.get("last_activity", current_time)) if (datetime.now() - last_activity).days > 7: data["status"] = "inactive" except: pass def _update_pending_threads(self, threads: List[Dict]): """Update pending threads list.""" # Keep existing threads and add new ones existing_contents = {t["content"] for t in self.work_context["pending_threads"]} for thread in threads: if thread["content"] not in existing_contents: self.work_context["pending_threads"].append(thread) # Sort and limit self.work_context["pending_threads"].sort( key=lambda x: x["timestamp"], reverse=True ) self.work_context["pending_threads"] = self.work_context["pending_threads"][:100] def _update_recent_topics(self, topics: List[str]): """Update recent topics list.""" # Merge with existing topics existing_topics = self.work_context["recent_topics"] # Add new topics at the beginning for topic in topics: if topic not in existing_topics: existing_topics.insert(0, topic) # Keep top 50 topics self.work_context["recent_topics"] = existing_topics[:50] def _analyze_priorities(self) -> List[Dict[str, Any]]: """Analyze and determine current priorities.""" priorities = [] # Priority from active projects for name, data in self.work_context["active_projects"].items(): if data["status"] in ["active", "highly_active"]: priorities.append({ "type": "project", "name": name, "reason": f"Active project with {data['total_activity']} recent activities", "urgency": "high" if data["status"] == "highly_active" else "medium" }) # Priority from pending threads critical_threads = [t for t in self.work_context["pending_threads"] if t["priority"] == "critical"] high_threads = [t for t in self.work_context["pending_threads"] if t["priority"] == "high"] for thread in critical_threads[:3]: priorities.append({ "type": "thread", "name": f"{thread['type']}: {thread['content'][:50]}...", "reason": "Critical pending item", "urgency": "critical" }) for thread in high_threads[:5]: priorities.append({ "type": "thread", "name": f"{thread['type']}: {thread['content'][:50]}...", "reason": "High priority pending item", "urgency": "high" }) # Sort by urgency urgency_order = {"critical": 0, "high": 1, "medium": 2, "low": 3} priorities.sort(key=lambda x: urgency_order.get(x["urgency"], 3)) return priorities[:10] # Top 10 priorities def _calculate_momentum_indicators(self) -> Dict[str, Any]: """Calculate work momentum indicators.""" indicators = { "active_project_count": len([p for p in self.work_context["active_projects"].values() if p["status"] != "inactive"]), "pending_thread_count": len(self.work_context["pending_threads"]), "critical_items": len([t for t in self.work_context["pending_threads"] if t["priority"] == "critical"]), "momentum_score": 0.0 } # Calculate momentum score score = 0.0 # Active projects boost momentum score += indicators["active_project_count"] * 10 # Pending threads reduce momentum (but some is good) if indicators["pending_thread_count"] < 10: score += 5 elif indicators["pending_thread_count"] < 20: score += 2 else: score -= (indicators["pending_thread_count"] - 20) * 0.5 # Critical items significantly impact momentum score -= indicators["critical_items"] * 5 indicators["momentum_score"] = max(0, score) indicators["momentum_status"] = self._assess_momentum_status(score) return indicators def _assess_momentum_status(self, score: float) -> str: """Assess momentum status based on score.""" if score >= 50: return "excellent" elif score >= 30: return "good" elif score >= 15: return "moderate" elif score >= 5: return "low" else: return "stalled" def _generate_recommendations(self) -> List[str]: """Generate work recommendations based on context.""" recommendations = [] momentum = self._calculate_momentum_indicators() # Recommendations based on momentum if momentum["momentum_status"] == "stalled": recommendations.append("🚨 Work momentum is stalled - focus on completing pending critical items") elif momentum["momentum_status"] == "low": recommendations.append("⚡ Boost momentum by tackling high-priority threads") # Recommendations based on pending items if momentum["critical_items"] > 0: recommendations.append(f"🔴 Address {momentum['critical_items']} critical items immediately") if momentum["pending_thread_count"] > 30: recommendations.append("📋 Too many pending threads - prioritize and defer non-essential items") # Project-specific recommendations highly_active = [name for name, data in self.work_context["active_projects"].items() if data["status"] == "highly_active"] if highly_active: recommendations.append(f"🎯 Main focus: {', '.join(highly_active)}") # Time-based recommendations inactive_projects = [name for name, data in self.work_context["active_projects"].items() if data["status"] == "inactive"] if inactive_projects: recommendations.append(f"🔄 Consider reactivating or archiving: {', '.join(inactive_projects)}") # Work pattern recommendations recent_topics = self.work_context["recent_topics"][:5] if recent_topics: recommendations.append(f"🧠 Recent focus areas: {', '.join(recent_topics)}") return recommendations def get_session_summary(self) -> Dict[str, Any]: """Get a summary suitable for session handoff.""" return { "active_projects": [ {"name": name, "status": data["status"], "last_activity": data.get("last_activity")} for name, data in self.work_context["active_projects"].items() if data["status"] != "inactive" ], "top_priorities": self._analyze_priorities()[:5], "pending_critical": [ t for t in self.work_context["pending_threads"] if t["priority"] == "critical" ][:3], "momentum_status": self._calculate_momentum_indicators()["momentum_status"], "quick_recommendations": self._generate_recommendations()[:3] } def mark_thread_resolved(self, thread_content: str): """Mark a pending thread as resolved.""" self.work_context["pending_threads"] = [ t for t in self.work_context["pending_threads"] if thread_content not in t["content"] ] self._save_work_context() def add_priority_item(self, item: str, urgency: str = "medium"): """Add a new priority item.""" self.work_context["pending_threads"].insert(0, { "type": "manual", "content": item, "timestamp": datetime.now().isoformat(), "source_role": "user", "priority": urgency }) self._save_work_context() # Module-level instance _work_context_instance = None def get_work_context_intelligence() -> WorkContextIntelligence: """Get or create the work context intelligence instance.""" global _work_context_instance if _work_context_instance is None: _work_context_instance = WorkContextIntelligence() return _work_context_instance