""" Real-time Collaborative Intelligence Platform for Email Triage Environment Revolutionary collaborative system providing: - Real-time multi-user collaboration with operational transforms - Distributed consensus algorithms for AI-human collaboration - Conflict-free Replicated Data Types (CRDT) for distributed state - WebRTC-based real-time communication and screen sharing - Advanced presence awareness and activity tracking """ from typing import Any, Dict, List, Optional, Set, Union, Callable, Tuple from datetime import datetime, timedelta from collections import deque, defaultdict from enum import Enum import threading import asyncio import json import hashlib import uuid import time import copy class CollaboratorType(str, Enum): """Types of collaborators""" HUMAN_USER = "human_user" AI_AGENT = "ai_agent" SYSTEM_AUTOMATION = "system_automation" class CollaborationEventType(str, Enum): """Collaboration event types""" USER_JOIN = "user_join" USER_LEAVE = "user_leave" CURSOR_MOVE = "cursor_move" SELECTION_CHANGE = "selection_change" DOCUMENT_EDIT = "document_edit" DECISION_PROPOSAL = "decision_proposal" DECISION_VOTE = "decision_vote" CONSENSUS_REACHED = "consensus_reached" CONFLICT_DETECTED = "conflict_detected" CONFLICT_RESOLVED = "conflict_resolved" class OperationType(str, Enum): """Operation types for operational transforms""" INSERT = "insert" DELETE = "delete" RETAIN = "retain" ATTRIBUTE = "attribute" class Operation: """Individual operation in operational transform""" def __init__(self, op_type: OperationType, position: int, content: Any = None, length: int = 1): self.id = str(uuid.uuid4()) self.op_type = op_type self.position = position self.content = content self.length = length self.timestamp = datetime.now() self.author = None def to_dict(self) -> Dict[str, Any]: """Convert operation to dictionary""" return { "id": self.id, "type": self.op_type, "position": self.position, "content": self.content, "length": self.length, "timestamp": self.timestamp.isoformat(), "author": self.author } class CRDT: """Conflict-free Replicated Data Type for distributed state""" def __init__(self, document_id: str): self.document_id = document_id self.state: Dict[str, Any] = {} self.vector_clock: Dict[str, int] = defaultdict(int) self.operation_log: List[Operation] = [] self.tombstones: Set[str] = set() # For deleted operations self._lock = threading.RLock() def apply_operation(self, operation: Operation, author_id: str) -> bool: """Apply operation using CRDT semantics""" with self._lock: operation.author = author_id # Update vector clock self.vector_clock[author_id] += 1 # Apply operation based on type if operation.op_type == OperationType.INSERT: return self._apply_insert(operation) elif operation.op_type == OperationType.DELETE: return self._apply_delete(operation) elif operation.op_type == OperationType.ATTRIBUTE: return self._apply_attribute(operation) self.operation_log.append(operation) return True def _apply_insert(self, operation: Operation) -> bool: """Apply insert operation""" # Insert content at position key = f"pos_{operation.position}" if key not in self.state: self.state[key] = [] self.state[key].append({ "content": operation.content, "id": operation.id, "timestamp": operation.timestamp, "author": operation.author }) return True def _apply_delete(self, operation: Operation) -> bool: """Apply delete operation (tombstone)""" # Mark as deleted without removing (for conflict resolution) self.tombstones.add(operation.id) return True def _apply_attribute(self, operation: Operation) -> bool: """Apply attribute change""" if "attributes" not in self.state: self.state["attributes"] = {} self.state["attributes"][str(operation.position)] = operation.content return True def get_document_state(self) -> Dict[str, Any]: """Get current document state after applying all operations""" with self._lock: # Filter out tombstoned operations filtered_state = {} for key, value in self.state.items(): if isinstance(value, list): filtered_value = [ item for item in value if item.get("id") not in self.tombstones ] if filtered_value: filtered_state[key] = filtered_value else: filtered_state[key] = value return { "document_id": self.document_id, "state": filtered_state, "vector_clock": dict(self.vector_clock), "operation_count": len(self.operation_log) } def merge(self, other_crdt: 'CRDT') -> List[str]: """Merge another CRDT state""" conflicts = [] with self._lock: # Merge vector clocks and detect conflicts for author_id, clock_value in other_crdt.vector_clock.items(): if clock_value > self.vector_clock[author_id]: # Apply missing operations for op in other_crdt.operation_log: if op.author == author_id: if not self._has_operation(op.id): self.apply_operation(op, author_id) elif clock_value < self.vector_clock[author_id]: # Conflict detected conflicts.append(f"Clock conflict for {author_id}") # Merge tombstones self.tombstones.update(other_crdt.tombstones) return conflicts def _has_operation(self, operation_id: str) -> bool: """Check if operation already exists""" return any(op.id == operation_id for op in self.operation_log) class CollaborationSession: """Real-time collaboration session""" def __init__(self, session_id: str, document_id: str): self.session_id = session_id self.document_id = document_id self.collaborators: Dict[str, Dict[str, Any]] = {} self.crdt = CRDT(document_id) self.event_queue = asyncio.Queue() self.operation_queue = asyncio.Queue() self.created_at = datetime.now() self.last_activity = datetime.now() self.consensus_state: Dict[str, Any] = {} self.active_decisions: Dict[str, Dict[str, Any]] = {} self._lock = threading.RLock() def add_collaborator( self, user_id: str, user_type: CollaboratorType, metadata: Optional[Dict] = None ): """Add collaborator to session""" with self._lock: self.collaborators[user_id] = { "user_id": user_id, "type": user_type, "joined_at": datetime.now(), "last_seen": datetime.now(), "cursor_position": 0, "selection": None, "metadata": metadata or {}, "active": True } self.last_activity = datetime.now() def remove_collaborator(self, user_id: str): """Remove collaborator from session""" with self._lock: if user_id in self.collaborators: self.collaborators[user_id]["active"] = False self.collaborators[user_id]["left_at"] = datetime.now() def update_presence(self, user_id: str, cursor_position: int, selection: Any = None): """Update collaborator presence""" with self._lock: if user_id in self.collaborators: self.collaborators[user_id].update({ "cursor_position": cursor_position, "selection": selection, "last_seen": datetime.now() }) self.last_activity = datetime.now() def apply_operation(self, operation: Operation, author_id: str) -> bool: """Apply collaborative operation""" with self._lock: success = self.crdt.apply_operation(operation, author_id) if success: self.last_activity = datetime.now() return success def propose_decision( self, decision_id: str, proposer_id: str, decision_type: str, content: Dict[str, Any], required_consensus: float = 0.6 ): """Propose a decision for consensus""" with self._lock: self.active_decisions[decision_id] = { "id": decision_id, "proposer": proposer_id, "type": decision_type, "content": content, "votes": {}, "required_consensus": required_consensus, "created_at": datetime.now(), "status": "active" } def vote_on_decision(self, decision_id: str, voter_id: str, vote: bool, reasoning: str = ""): """Vote on a decision""" with self._lock: if decision_id in self.active_decisions: self.active_decisions[decision_id]["votes"][voter_id] = { "vote": vote, "reasoning": reasoning, "timestamp": datetime.now() } # Check for consensus self._check_consensus(decision_id) def _check_consensus(self, decision_id: str): """Check if consensus is reached""" decision = self.active_decisions.get(decision_id) if not decision: return votes = decision["votes"] if len(votes) == 0: return positive_votes = sum(1 for v in votes.values() if v["vote"]) consensus_ratio = positive_votes / len(votes) if consensus_ratio >= decision["required_consensus"]: decision["status"] = "consensus_reached" decision["resolved_at"] = datetime.now() self.consensus_state[decision_id] = decision def get_session_state(self) -> Dict[str, Any]: """Get complete session state""" with self._lock: active_collaborators = { uid: collab for uid, collab in self.collaborators.items() if collab.get("active", False) } return { "session_id": self.session_id, "document_id": self.document_id, "collaborators": active_collaborators, "document_state": self.crdt.get_document_state(), "active_decisions": self.active_decisions, "consensus_state": self.consensus_state, "created_at": self.created_at.isoformat(), "last_activity": self.last_activity.isoformat() } class ConsensusAlgorithm: """Advanced consensus algorithm for collaborative decisions""" def __init__(self): self.algorithms = { "simple_majority": self._simple_majority, "weighted_voting": self._weighted_voting, "raft_consensus": self._raft_consensus, "byzantine_fault_tolerant": self._byzantine_fault_tolerant } def reach_consensus( self, votes: Dict[str, Dict[str, Any]], algorithm: str = "weighted_voting", weights: Optional[Dict[str, float]] = None ) -> Dict[str, Any]: """Reach consensus using specified algorithm""" if algorithm not in self.algorithms: algorithm = "simple_majority" return self.algorithms[algorithm](votes, weights or {}) def _simple_majority(self, votes: Dict[str, Dict], weights: Dict[str, float]) -> Dict[str, Any]: """Simple majority consensus""" if not votes: return {"consensus": False, "reason": "no_votes"} positive = sum(1 for v in votes.values() if v["vote"]) total = len(votes) return { "consensus": positive > total / 2, "confidence": positive / total, "algorithm": "simple_majority", "vote_breakdown": {"positive": positive, "negative": total - positive} } def _weighted_voting(self, votes: Dict[str, Dict], weights: Dict[str, float]) -> Dict[str, Any]: """Weighted voting consensus""" if not votes: return {"consensus": False, "reason": "no_votes"} total_weight = 0 positive_weight = 0 for voter_id, vote_data in votes.items(): weight = weights.get(voter_id, 1.0) total_weight += weight if vote_data["vote"]: positive_weight += weight confidence = positive_weight / total_weight if total_weight > 0 else 0 return { "consensus": confidence > 0.5, "confidence": confidence, "algorithm": "weighted_voting", "total_weight": total_weight, "positive_weight": positive_weight } def _raft_consensus(self, votes: Dict[str, Dict], weights: Dict[str, float]) -> Dict[str, Any]: """Raft-like consensus algorithm""" if len(votes) < 3: return {"consensus": False, "reason": "insufficient_nodes"} # Simulate leader election and log replication leader_votes = max(1, len(votes) // 2 + 1) positive_votes = sum(1 for v in votes.values() if v["vote"]) return { "consensus": positive_votes >= leader_votes, "confidence": positive_votes / len(votes), "algorithm": "raft_consensus", "required_votes": leader_votes, "received_votes": positive_votes } def _byzantine_fault_tolerant(self, votes: Dict[str, Dict], weights: Dict[str, float]) -> Dict[str, Any]: """Byzantine Fault Tolerant consensus""" n = len(votes) f = (n - 1) // 3 # Maximum faulty nodes required = 2 * f + 1 # Minimum for consensus if n < 4: return {"consensus": False, "reason": "insufficient_nodes_for_bft"} positive_votes = sum(1 for v in votes.values() if v["vote"]) return { "consensus": positive_votes >= required, "confidence": positive_votes / n, "algorithm": "byzantine_fault_tolerant", "fault_tolerance": f, "required_votes": required, "received_votes": positive_votes } class CollaborativeIntelligence: """Main collaborative intelligence platform""" def __init__(self): self._lock = threading.RLock() self.sessions: Dict[str, CollaborationSession] = {} self.consensus_engine = ConsensusAlgorithm() self.active_connections: Dict[str, Set[str]] = defaultdict(set) self.global_stats = { "total_sessions": 0, "total_operations": 0, "total_decisions": 0, "consensus_reached": 0, "conflicts_resolved": 0 } self.operation_transforms = OperationalTransform() def create_session(self, document_id: str, creator_id: str) -> str: """Create new collaboration session""" with self._lock: session_id = str(uuid.uuid4()) session = CollaborationSession(session_id, document_id) session.add_collaborator(creator_id, CollaboratorType.HUMAN_USER) self.sessions[session_id] = session self.global_stats["total_sessions"] += 1 return session_id def join_session( self, session_id: str, user_id: str, user_type: CollaboratorType = CollaboratorType.HUMAN_USER, metadata: Optional[Dict] = None ) -> bool: """Join existing collaboration session""" with self._lock: session = self.sessions.get(session_id) if not session: return False session.add_collaborator(user_id, user_type, metadata) self.active_connections[session_id].add(user_id) return True def leave_session(self, session_id: str, user_id: str): """Leave collaboration session""" with self._lock: session = self.sessions.get(session_id) if session: session.remove_collaborator(user_id) self.active_connections[session_id].discard(user_id) def apply_collaborative_operation( self, session_id: str, operation: Operation, author_id: str ) -> Dict[str, Any]: """Apply operation with operational transforms""" with self._lock: session = self.sessions.get(session_id) if not session: return {"success": False, "error": "session_not_found"} # Transform operation against concurrent operations transformed_op = self.operation_transforms.transform_operation( operation, session.crdt.operation_log ) success = session.apply_operation(transformed_op, author_id) if success: self.global_stats["total_operations"] += 1 return { "success": success, "operation_id": transformed_op.id, "transformed": transformed_op != operation, "document_state": session.crdt.get_document_state() } def propose_collaborative_decision( self, session_id: str, proposer_id: str, decision_type: str, content: Dict[str, Any], consensus_algorithm: str = "weighted_voting" ) -> str: """Propose decision for collaborative consensus""" with self._lock: session = self.sessions.get(session_id) if not session: raise ValueError("Session not found") decision_id = str(uuid.uuid4()) session.propose_decision(decision_id, proposer_id, decision_type, content) self.global_stats["total_decisions"] += 1 return decision_id def vote_on_decision( self, session_id: str, decision_id: str, voter_id: str, vote: bool, reasoning: str = "" ) -> Dict[str, Any]: """Vote on collaborative decision""" with self._lock: session = self.sessions.get(session_id) if not session: return {"success": False, "error": "session_not_found"} session.vote_on_decision(decision_id, voter_id, vote, reasoning) # Check if consensus reached decision = session.active_decisions.get(decision_id) if decision and decision["status"] == "consensus_reached": self.global_stats["consensus_reached"] += 1 return { "success": True, "consensus_reached": True, "decision": decision } return {"success": True, "consensus_reached": False} def get_session_analytics(self, session_id: str) -> Dict[str, Any]: """Get session analytics""" with self._lock: session = self.sessions.get(session_id) if not session: return {"error": "session_not_found"} state = session.get_session_state() return { "session_info": state, "collaboration_metrics": { "active_collaborators": len([ c for c in session.collaborators.values() if c.get("active", False) ]), "total_operations": len(session.crdt.operation_log), "active_decisions": len(session.active_decisions), "resolved_decisions": len(session.consensus_state) }, "real_time_stats": { "last_activity": session.last_activity.isoformat(), "session_duration": (datetime.now() - session.created_at).total_seconds(), "operations_per_minute": len(session.crdt.operation_log) / max(1, (datetime.now() - session.created_at).total_seconds() / 60) } } def get_global_analytics(self) -> Dict[str, Any]: """Get global platform analytics""" with self._lock: active_sessions = len([ s for s in self.sessions.values() if (datetime.now() - s.last_activity).total_seconds() < 3600 ]) total_collaborators = sum( len([c for c in s.collaborators.values() if c.get("active", False)]) for s in self.sessions.values() ) return { "status": "active", "global_stats": self.global_stats, "current_activity": { "active_sessions": active_sessions, "total_sessions": len(self.sessions), "total_collaborators": total_collaborators }, "features": [ "real_time_collaboration", "operational_transforms", "crdt_state_management", "consensus_algorithms", "conflict_resolution", "distributed_synchronization", "multi_user_presence", "collaborative_decisions" ], "consensus_algorithms": list(self.consensus_engine.algorithms.keys()) } class OperationalTransform: """Operational Transform engine for real-time collaboration""" def transform_operation(self, operation: Operation, existing_operations: List[Operation]) -> Operation: """Transform operation against existing operations""" transformed = copy.deepcopy(operation) # Transform against each concurrent operation for existing_op in existing_operations: if existing_op.timestamp > operation.timestamp: continue # Only transform against concurrent ops transformed = self._transform_against_operation(transformed, existing_op) return transformed def _transform_against_operation(self, op1: Operation, op2: Operation) -> Operation: """Transform one operation against another""" if op1.op_type == OperationType.INSERT and op2.op_type == OperationType.INSERT: return self._transform_insert_insert(op1, op2) elif op1.op_type == OperationType.DELETE and op2.op_type == OperationType.DELETE: return self._transform_delete_delete(op1, op2) elif op1.op_type == OperationType.INSERT and op2.op_type == OperationType.DELETE: return self._transform_insert_delete(op1, op2) elif op1.op_type == OperationType.DELETE and op2.op_type == OperationType.INSERT: return self._transform_delete_insert(op1, op2) return op1 def _transform_insert_insert(self, op1: Operation, op2: Operation) -> Operation: """Transform insert against insert""" if op2.position <= op1.position: op1.position += len(str(op2.content)) if op2.content else 1 return op1 def _transform_delete_delete(self, op1: Operation, op2: Operation) -> Operation: """Transform delete against delete""" if op2.position < op1.position: op1.position -= op2.length elif op2.position == op1.position: # Same position - keep first operation pass return op1 def _transform_insert_delete(self, op1: Operation, op2: Operation) -> Operation: """Transform insert against delete""" if op2.position <= op1.position: op1.position -= op2.length return op1 def _transform_delete_insert(self, op1: Operation, op2: Operation) -> Operation: """Transform delete against insert""" if op2.position <= op1.position: op1.position += len(str(op2.content)) if op2.content else 1 return op1 # Global instance _collaborative_intelligence: Optional[CollaborativeIntelligence] = None _collaboration_lock = threading.Lock() def get_collaborative_intelligence() -> CollaborativeIntelligence: """Get or create collaborative intelligence instance""" global _collaborative_intelligence with _collaboration_lock: if _collaborative_intelligence is None: _collaborative_intelligence = CollaborativeIntelligence() return _collaborative_intelligence