# Copyright (c) 2025 Patrick Motsch # All rights reserved. """ Interface to Chatbot database and AI Connectors. Uses the PostgreSQL connector for data access with user/feature-instance filtering. Chatbot-specific models in poweron_chatbot (separate from workflow engine). """ import logging import uuid import math from typing import Dict, Any, List, Optional, Union from enum import Enum from pydantic import BaseModel, Field import asyncio from modules.security.rbac import RbacClass from modules.datamodels.datamodelRbac import AccessRuleContext from modules.datamodels.datamodelUam import AccessLevel from modules.datamodels.datamodelChat import UserInputRequest from modules.shared.timeUtils import getUtcTimestamp, parseTimestamp # ============================================================================= # Chatbot-specific Pydantic models for poweron_chatbot (per-instance isolation) # ============================================================================= class ChatbotDocument(BaseModel): """Documents attached to chatbot messages.""" id: str = Field(default_factory=lambda: str(uuid.uuid4()), description="Primary key") messageId: str = Field(description="Foreign key to message") fileId: str = Field(description="Foreign key to file") fileName: str = Field(description="Name of the file") fileSize: int = Field(description="Size of the file") mimeType: str = Field(description="MIME type of the file") roundNumber: Optional[int] = Field(None, description="Round number in workflow") taskNumber: Optional[int] = Field(None, description="Task number within round") actionNumber: Optional[int] = Field(None, description="Action number within task") actionId: Optional[str] = Field(None, description="ID of the action that created this document") class ChatbotMessage(BaseModel): """Messages in chatbot conversations. Must match bridge format in memory.py.""" id: str = Field(default_factory=lambda: str(uuid.uuid4()), description="Primary key") conversationId: str = Field(description="Foreign key to conversation") parentMessageId: Optional[str] = Field(None, description="Parent message ID for threading") documents: List[ChatbotDocument] = Field(default_factory=list, description="Associated documents") documentsLabel: Optional[str] = Field(None, description="Label for the set of documents") message: Optional[str] = Field(None, description="Message content") role: str = Field(description="Role of the message sender") status: str = Field(description="Status of the message (first, step, last)") sequenceNr: Optional[int] = Field(default=0, description="Sequence number of the message") publishedAt: Optional[float] = Field(default=None, description="When the message was published (UTC timestamp)") success: Optional[bool] = Field(None, description="Whether the message processing was successful") actionId: Optional[str] = Field(None, description="ID of the action that produced this message") actionMethod: Optional[str] = Field(None, description="Method of the action that produced this message") actionName: Optional[str] = Field(None, description="Name of the action that produced this message") roundNumber: Optional[int] = Field(None, description="Round number in workflow") taskNumber: Optional[int] = Field(None, description="Task number within round") actionNumber: Optional[int] = Field(None, description="Action number within task") taskProgress: Optional[str] = Field(None, description="Task progress status") actionProgress: Optional[str] = Field(None, description="Action progress status") class ChatbotLog(BaseModel): """Log entries for chatbot conversations.""" id: str = Field(default_factory=lambda: str(uuid.uuid4()), description="Primary key") conversationId: str = Field(description="Foreign key to conversation") message: str = Field(description="Log message") type: str = Field(description="Log type (info, warning, error, etc.)") timestamp: float = Field(default_factory=getUtcTimestamp, description="When the log entry was created (UTC timestamp)") status: Optional[str] = Field(None, description="Status of the log entry") progress: Optional[float] = Field(None, description="Progress indicator (0.0 to 1.0)") performance: Optional[Dict[str, Any]] = Field(None, description="Performance metrics") parentId: Optional[str] = Field(None, description="Parent operation ID") operationId: Optional[str] = Field(None, description="Operation ID to group related log entries") roundNumber: Optional[int] = Field(None, description="Round number in workflow") taskNumber: Optional[int] = Field(None, description="Task number within round") actionNumber: Optional[int] = Field(None, description="Action number within task") class ChatbotWorkflowModeEnum(str, Enum): WORKFLOW_CHATBOT = "Chatbot" class ChatbotConversation(BaseModel): """Chatbot conversation container. Per feature-instance isolation via featureInstanceId.""" id: str = Field(default_factory=lambda: str(uuid.uuid4()), description="Primary key") featureInstanceId: str = Field(description="Feature instance ID for per-instance isolation") name: Optional[str] = Field(None, description="Name of the conversation") status: str = Field(default="running", description="Current status") currentRound: int = Field(default=0, description="Current round number") lastActivity: float = Field(default_factory=getUtcTimestamp, description="Timestamp of last activity") startedAt: float = Field(default_factory=getUtcTimestamp, description="When the conversation started") workflowMode: ChatbotWorkflowModeEnum = Field(default=ChatbotWorkflowModeEnum.WORKFLOW_CHATBOT, description="Workflow mode") maxSteps: int = Field(default=10, description="Maximum number of iterations") # Hydrated from child tables (not stored in DB) logs: List[ChatbotLog] = Field(default_factory=list, description="Conversation logs") messages: List[ChatbotMessage] = Field(default_factory=list, description="Conversation messages") import json from modules.datamodels.datamodelUam import User # DYNAMIC PART: Connectors to the Interface from modules.connectors.connectorDbPostgre import DatabaseConnector from modules.datamodels.datamodelPagination import PaginationParams, PaginatedResult from modules.interfaces.interfaceRbac import getRecordsetWithRBAC # Basic Configurations from modules.shared.configuration import APP_CONFIG logger = logging.getLogger(__name__) # Singleton factory for Chat instances _chatInterfaces = {} def storeDebugMessageAndDocuments(message, currentUser, mandateId=None, featureInstanceId=None) -> None: """ Store message and documents (metadata and file bytes) for debugging purposes. Structure: {log_dir}/debug/messages/m_round_task_action_timestamp/documentlist_label/ - message.json, message_text.txt - document_###_metadata.json - document_###_ (actual file bytes) Args: message: ChatbotMessage or ChatMessage-like object to store currentUser: Current user for component interface access mandateId: Mandate ID for RBAC context (avoids overwriting singleton state) featureInstanceId: Feature instance ID for RBAC context """ try: import os from datetime import datetime, UTC from modules.shared.debugLogger import _getBaseDebugDir, _ensureDir from modules.interfaces.interfaceDbManagement import getInterface # Create base debug directory (use base debug dir, not prompts subdirectory) baseDebugDir = _getBaseDebugDir() debug_root = os.path.join(baseDebugDir, 'messages') _ensureDir(debug_root) # Generate timestamp timestamp = datetime.now(UTC).strftime('%Y%m%d-%H%M%S-%f')[:-3] # Create message folder name: m_round_task_action_timestamp # Use actual values from message, not defaults round_str = str(message.roundNumber) if message.roundNumber is not None else "0" task_str = str(message.taskNumber) if message.taskNumber is not None else "0" action_str = str(message.actionNumber) if message.actionNumber is not None else "0" message_folder = f"{timestamp}_m_{round_str}_{task_str}_{action_str}" message_path = os.path.join(debug_root, message_folder) os.makedirs(message_path, exist_ok=True) # Store message data - use dict() instead of model_dump() for compatibility message_file = os.path.join(message_path, "message.json") with open(message_file, "w", encoding="utf-8") as f: # Convert message to dict manually to avoid model_dump() issues message_dict = { "id": message.id, "workflowId": getattr(message, "conversationId", None) or getattr(message, "workflowId", ""), "parentMessageId": message.parentMessageId, "message": message.message, "role": message.role, "status": message.status, "sequenceNr": message.sequenceNr, "publishedAt": message.publishedAt, "roundNumber": message.roundNumber, "taskNumber": message.taskNumber, "actionNumber": message.actionNumber, "documentsLabel": message.documentsLabel, "actionId": message.actionId, "actionMethod": message.actionMethod, "actionName": message.actionName, "success": message.success, "documents": [] } json.dump(message_dict, f, indent=2, ensure_ascii=False, default=str) # Store message content as text if message.message: message_text_file = os.path.join(message_path, "message_text.txt") with open(message_text_file, "w", encoding="utf-8") as f: f.write(str(message.message)) # Store documents if provided if message.documents and len(message.documents) > 0: # Group documents by documentsLabel documents_by_label = {} for doc in message.documents: label = message.documentsLabel or 'default' if label not in documents_by_label: documents_by_label[label] = [] documents_by_label[label].append(doc) # Create subfolder for each document label for label, docs in documents_by_label.items(): # Sanitize label for filesystem safe_label = "".join(c for c in str(label) if c.isalnum() or c in (' ', '-', '_')).rstrip() safe_label = safe_label.replace(' ', '_') if not safe_label: safe_label = "default" label_folder = os.path.join(message_path, safe_label) _ensureDir(label_folder) # Store each document for i, doc in enumerate(docs): # Create document metadata file doc_meta = { "id": doc.id, "messageId": doc.messageId, "fileId": doc.fileId, "fileName": doc.fileName, "fileSize": doc.fileSize, "mimeType": doc.mimeType, "roundNumber": doc.roundNumber, "taskNumber": doc.taskNumber, "actionNumber": doc.actionNumber, "actionId": doc.actionId } doc_meta_file = os.path.join(label_folder, f"document_{i+1:03d}_metadata.json") with open(doc_meta_file, "w", encoding="utf-8") as f: json.dump(doc_meta, f, indent=2, ensure_ascii=False, default=str) # Also store the actual file bytes next to metadata for debugging try: componentInterface = getInterface(currentUser, mandateId=mandateId, featureInstanceId=featureInstanceId) file_bytes = componentInterface.getFileData(doc.fileId) if file_bytes: # Build a safe filename preserving original name safe_name = doc.fileName or f"document_{i+1:03d}" # Avoid path traversal safe_name = os.path.basename(safe_name) doc_file_path = os.path.join(label_folder, f"document_{i+1:03d}_" + safe_name) with open(doc_file_path, "wb") as df: df.write(file_bytes) else: pass except Exception as e: pass except Exception as e: # Silent fail - don't break main flow pass class ChatObjects: """ Interface to Chat database and AI Connectors. Uses the JSON connector for data access with added language support. """ def __init__(self, currentUser: Optional[User] = None, mandateId: Optional[str] = None, featureInstanceId: Optional[str] = None): """Initializes the Chat Interface. Args: currentUser: The authenticated user mandateId: The mandate ID from RequestContext (X-Mandate-Id header) featureInstanceId: The feature instance ID from RequestContext (X-Feature-Instance-Id header) """ # Initialize variables self.currentUser = currentUser # Store User object directly self.userId = currentUser.id if currentUser else None # Use mandateId from parameter (Request-Context), not from user object self.mandateId = mandateId self.featureInstanceId = featureInstanceId self.featureCode = "chatbot" # For RBAC buildDataObjectKey self.rbac = None # RBAC interface # Initialize services self._initializeServices() # Initialize database self._initializeDatabase() # Set user context if provided if currentUser: self.setUserContext(currentUser, mandateId=mandateId, featureInstanceId=featureInstanceId) # ===== Generic Utility Methods ===== def _isObjectField(self, fieldType) -> bool: """Check if a field type represents a complex object (not a simple type).""" # Simple scalar types if fieldType in (str, int, float, bool, type(None)): return False # Everything else is an object return True def _separateObjectFields(self, model_class, data: Dict[str, Any]) -> tuple[Dict[str, Any], Dict[str, Any]]: """Separate simple fields from object fields based on Pydantic model structure.""" simpleFields = {} objectFields = {} # Get field information from the Pydantic model modelFields = model_class.model_fields for fieldName, value in data.items(): # Check if this field should be stored as JSONB in the database if fieldName in modelFields: fieldInfo = modelFields[fieldName] # Pydantic v2 only fieldType = fieldInfo.annotation # Always route relational/object fields to object_fields for separate handling # These fields are stored in separate normalized tables, not as JSONB if fieldName in ['documents', 'stats', 'logs', 'messages']: objectFields[fieldName] = value continue # Check if this is a JSONB field (Dict, List, or complex types) # Purely type-based detection - no hardcoded field names if (fieldType == dict or fieldType == list or (hasattr(fieldType, '__origin__') and fieldType.__origin__ in (dict, list))): # Store as JSONB - include in simple_fields for database storage simpleFields[fieldName] = value elif isinstance(value, (str, int, float, bool, type(None))): # Simple scalar types simpleFields[fieldName] = value else: # Complex objects that should be filtered out objectFields[fieldName] = value else: # Field not in model - treat as scalar if simple, otherwise filter out # BUT: always include metadata fields (_createdBy, _createdAt, etc.) as they're handled by connector if fieldName.startswith("_"): # Metadata fields should be passed through to connector simpleFields[fieldName] = value elif isinstance(value, (str, int, float, bool, type(None))): simpleFields[fieldName] = value else: objectFields[fieldName] = value return simpleFields, objectFields def _initializeServices(self): pass def setUserContext(self, currentUser: User, mandateId: Optional[str] = None, featureInstanceId: Optional[str] = None): """Sets the user context for the interface. Args: currentUser: The authenticated user mandateId: The mandate ID from RequestContext (X-Mandate-Id header) featureInstanceId: The feature instance ID from RequestContext (X-Feature-Instance-Id header) """ self.currentUser = currentUser # Store User object directly self.userId = currentUser.id # Use mandateId from parameter (Request-Context), not from user object self.mandateId = mandateId self.featureInstanceId = featureInstanceId if not self.userId: raise ValueError("Invalid user context: id is required") # Note: mandateId is ALWAYS optional here - it comes from Request-Context, not from User. # Users are NOT assigned to mandates by design - they get mandate context from the request. # sysAdmin users can additionally perform cross-mandate operations. # Without mandateId, operations will be filtered to accessible mandates via RBAC. # Add language settings self.userLanguage = currentUser.language # Default user language # Initialize RBAC interface if not self.currentUser: raise ValueError("User context is required for RBAC") # Get DbApp connection for RBAC AccessRule queries from modules.security.rootAccess import getRootDbAppConnector dbApp = getRootDbAppConnector() self.rbac = RbacClass(self.db, dbApp=dbApp) # Update database context self.db.updateContext(self.userId) def __del__(self): """Cleanup method to close database connection.""" if hasattr(self, 'db') and self.db is not None: try: self.db.close() except Exception as e: logger.error(f"Error closing database connection: {e}") def _initializeDatabase(self): """Initializes the database connection directly.""" try: # Get configuration values with defaults dbHost = APP_CONFIG.get("DB_HOST", "_no_config_default_data") dbDatabase = "poweron_chatbot" dbUser = APP_CONFIG.get("DB_USER") dbPassword = APP_CONFIG.get("DB_PASSWORD_SECRET") dbPort = int(APP_CONFIG.get("DB_PORT", 5432)) from modules.connectors.connectorDbPostgre import _get_cached_connector self.db = _get_cached_connector( dbHost=dbHost, dbDatabase=dbDatabase, dbUser=dbUser, dbPassword=dbPassword, dbPort=dbPort, userId=self.userId ) logger.info("Database initialized successfully") except Exception as e: logger.error(f"Failed to initialize database: {str(e)}") raise def _initRecords(self): """Initializes standard records in the database if they don't exist.""" pass def checkRbacPermission( self, modelClass: type, operation: str, recordId: Optional[str] = None ) -> bool: """ Check RBAC permission for a specific operation on a table. Args: modelClass: Pydantic model class for the table operation: Operation to check ('create', 'update', 'delete', 'read') recordId: Optional record ID for specific record check Returns: Boolean indicating permission """ if not self.rbac or not self.currentUser: return False tableName = modelClass.__name__ from modules.interfaces.interfaceRbac import buildDataObjectKey objectKey = buildDataObjectKey(tableName, featureCode=getattr(self, 'featureCode', 'chatbot')) permissions = self.rbac.getUserPermissions( self.currentUser, AccessRuleContext.DATA, objectKey, mandateId=self.mandateId, featureInstanceId=self.featureInstanceId ) if operation == "create": return permissions.create != AccessLevel.NONE elif operation == "update": return permissions.update != AccessLevel.NONE elif operation == "delete": return permissions.delete != AccessLevel.NONE elif operation == "read": return permissions.read != AccessLevel.NONE else: return False def _applyFilters(self, records: List[Dict[str, Any]], filters: Dict[str, Any]) -> List[Dict[str, Any]]: """ Apply filter criteria to records. Supports: - General search: {"search": "text"} - searches across all text fields - Field-specific filters: - Simple: {"status": "running"} - equals match - With operator: {"status": {"operator": "equals", "value": "running"}} - Operators: equals, contains, gt, gte, lt, lte, in, notIn, startsWith, endsWith Args: records: List of record dictionaries to filter filters: Filter criteria dictionary Returns: Filtered list of records """ if not filters or not records: return records filtered = [] for record in records: matches = True # Handle general search across text fields if "search" in filters: search_term = str(filters["search"]).lower() if search_term: # Search in all string fields found = False for key, value in record.items(): if isinstance(value, str) and search_term in value.lower(): found = True break elif isinstance(value, (int, float)) and search_term in str(value): found = True break if not found: matches = False # Handle field-specific filters for field_name, filter_value in filters.items(): if field_name == "search": continue # Already handled above if field_name not in record: matches = False break record_value = record.get(field_name) # Handle simple value (equals operator) if not isinstance(filter_value, dict): if record_value != filter_value: matches = False break continue # Handle filter with operator operator = filter_value.get("operator", "equals") filter_val = filter_value.get("value") if operator in ["equals", "eq"]: if record_value != filter_val: matches = False break elif operator == "contains": record_str = str(record_value).lower() if record_value is not None else "" filter_str = str(filter_val).lower() if filter_val is not None else "" if filter_str not in record_str: matches = False break elif operator == "startsWith": record_str = str(record_value).lower() if record_value is not None else "" filter_str = str(filter_val).lower() if filter_val is not None else "" if not record_str.startswith(filter_str): matches = False break elif operator == "endsWith": record_str = str(record_value).lower() if record_value is not None else "" filter_str = str(filter_val).lower() if filter_val is not None else "" if not record_str.endswith(filter_str): matches = False break elif operator == "gt": try: record_num = float(record_value) if record_value is not None else float('-inf') filter_num = float(filter_val) if filter_val is not None else float('-inf') if record_num <= filter_num: matches = False break except (ValueError, TypeError): matches = False break elif operator == "gte": try: record_num = float(record_value) if record_value is not None else float('-inf') filter_num = float(filter_val) if filter_val is not None else float('-inf') if record_num < filter_num: matches = False break except (ValueError, TypeError): matches = False break elif operator == "lt": try: record_num = float(record_value) if record_value is not None else float('inf') filter_num = float(filter_val) if filter_val is not None else float('inf') if record_num >= filter_num: matches = False break except (ValueError, TypeError): matches = False break elif operator == "lte": try: record_num = float(record_value) if record_value is not None else float('inf') filter_num = float(filter_val) if filter_val is not None else float('inf') if record_num > filter_num: matches = False break except (ValueError, TypeError): matches = False break elif operator == "in": if not isinstance(filter_val, list): filter_val = [filter_val] if record_value not in filter_val: matches = False break elif operator == "notIn": if not isinstance(filter_val, list): filter_val = [filter_val] if record_value in filter_val: matches = False break else: # Unknown operator - default to equals if record_value != filter_val: matches = False break if matches: filtered.append(record) return filtered def _applySorting(self, records: List[Dict[str, Any]], sortFields: List[Any]) -> List[Dict[str, Any]]: """Apply multi-level sorting to records using stable sort (sorts from least to most significant field).""" if not sortFields: return records # Start with a copy to avoid modifying original sortedRecords = list(records) # Sort from least significant to most significant field (reverse order) # Python's sort is stable, so this creates proper multi-level sorting for sortField in reversed(sortFields): # Handle both dict and object formats if isinstance(sortField, dict): fieldName = sortField.get("field") direction = sortField.get("direction", "asc") else: fieldName = getattr(sortField, "field", None) direction = getattr(sortField, "direction", "asc") if not fieldName: continue isDesc = (direction == "desc") def sortKey(record): value = record.get(fieldName) # Handle None values - place them at the end for both directions if value is None: # Use a special value that sorts last return (1, "") # (is_none_flag, empty_value) - sorts after (0, ...) else: # Return tuple with type indicator for proper comparison if isinstance(value, (int, float)): return (0, value) elif isinstance(value, str): return (0, value) elif isinstance(value, bool): return (0, value) else: return (0, str(value)) # Sort with reverse parameter for descending sortedRecords.sort(key=sortKey, reverse=isDesc) return sortedRecords # Utilities def getInitialId(self, model_class: type) -> Optional[str]: """Returns the initial ID for a table.""" return self.db.getInitialId(model_class) # Conversation methods (chatbot-specific, poweron_chatbot) def getConversations(self, pagination: Optional[PaginationParams] = None) -> Union[List[Dict[str, Any]], PaginatedResult]: """ Returns conversations for current feature instance based on user access. Supports optional pagination, sorting, and filtering. Args: pagination: Optional pagination parameters. If None, returns all items. Returns: If pagination is None: List[Dict[str, Any]] If pagination is provided: PaginatedResult with items and metadata """ # Use RBAC filtering with featureInstanceId for instance-level isolation filteredConversations = getRecordsetWithRBAC(self.db, ChatbotConversation, self.currentUser, mandateId=self.mandateId, featureInstanceId=self.featureInstanceId, featureCode="chatbot" ) # If no pagination requested, return all items (no sorting - frontend handles it) if pagination is None: return filteredConversations # Apply filtering (if filters provided) if pagination.filters: filteredConversations = self._applyFilters(filteredConversations, pagination.filters) # Apply sorting (in order of sortFields) - only if provided by frontend if pagination.sort: filteredConversations = self._applySorting(filteredConversations, pagination.sort) # Count total items after filters totalItems = len(filteredConversations) totalPages = math.ceil(totalItems / pagination.pageSize) if totalItems > 0 else 0 # Apply pagination (skip/limit) startIdx = (pagination.page - 1) * pagination.pageSize endIdx = startIdx + pagination.pageSize pagedConversations = filteredConversations[startIdx:endIdx] return PaginatedResult( items=pagedConversations, totalItems=totalItems, totalPages=totalPages ) def getWorkflows(self, pagination: Optional[PaginationParams] = None) -> Union[List[Dict[str, Any]], PaginatedResult]: """Backward-compat alias for getConversations.""" return self.getConversations(pagination) def getConversation(self, conversationId: str) -> Optional[ChatbotConversation]: """Returns a conversation by ID if user has access.""" # Use RBAC filtering with featureInstanceId for instance-level isolation conversations = getRecordsetWithRBAC(self.db, ChatbotConversation, self.currentUser, recordFilter={"id": conversationId}, mandateId=self.mandateId, featureInstanceId=self.featureInstanceId, featureCode="chatbot" ) if not conversations: return None conv = conversations[0] try: # Load related data from normalized tables logs = self.getLogs(conversationId) messages = self.getMessages(conversationId) # Build ChatbotConversation with hydrated logs/messages return ChatbotConversation( id=conv["id"], featureInstanceId=conv.get("featureInstanceId") or self.featureInstanceId or "", name=conv.get("name"), status=conv.get("status", "running"), currentRound=conv.get("currentRound", 0) or 0, lastActivity=conv.get("lastActivity", getUtcTimestamp()), startedAt=conv.get("startedAt", getUtcTimestamp()), workflowMode=ChatbotWorkflowModeEnum(conv.get("workflowMode", "Chatbot")), maxSteps=conv.get("maxSteps") if conv.get("maxSteps") is not None else 10, logs=logs, messages=messages ) except Exception as e: logger.error(f"Error validating conversation data: {str(e)}") return None def getWorkflow(self, workflowId: str) -> Optional[ChatbotConversation]: """Backward-compat alias: workflowId maps to conversationId.""" return self.getConversation(workflowId) def getWorkflowMinimal(self, workflowId: str) -> Optional[ChatbotConversation]: """Lightweight fetch: conversation record only, no logs/messages. For resume path.""" conversations = getRecordsetWithRBAC( self.db, ChatbotConversation, self.currentUser, recordFilter={"id": workflowId}, mandateId=self.mandateId, featureInstanceId=self.featureInstanceId, featureCode="chatbot", ) if not conversations: return None conv = conversations[0] max_steps = conv.get("maxSteps") return ChatbotConversation( id=conv["id"], featureInstanceId=conv.get("featureInstanceId") or self.featureInstanceId or "", name=conv.get("name"), status=conv.get("status", "running"), currentRound=conv.get("currentRound", 0) or 0, lastActivity=conv.get("lastActivity", getUtcTimestamp()), startedAt=conv.get("startedAt", getUtcTimestamp()), workflowMode=ChatbotWorkflowModeEnum(conv.get("workflowMode", "Chatbot")), maxSteps=max_steps if max_steps is not None else 10, logs=[], messages=[], ) def getMessageCount(self, conversationId: str) -> int: """Returns message count for a conversation (single query, no document fetch).""" messages = getRecordsetWithRBAC( self.db, ChatbotMessage, self.currentUser, recordFilter={"conversationId": conversationId}, mandateId=self.mandateId, featureInstanceId=self.featureInstanceId, featureCode="chatbot", ) return len(messages) if messages else 0 def updateWorkflowMinimal( self, workflowId: str, workflowData: Dict[str, Any] ) -> ChatbotConversation: """Lightweight update: no logs/messages fetch. For resume when caller has minimal workflow.""" if not self.checkRbacPermission(ChatbotConversation, "update", workflowId): raise PermissionError(f"No permission to update conversation {workflowId}") simpleFields, _ = self._separateObjectFields(ChatbotConversation, workflowData) simpleFields["lastActivity"] = getUtcTimestamp() updated = self.db.recordModify(ChatbotConversation, workflowId, simpleFields) max_steps = updated.get("maxSteps") return ChatbotConversation( id=updated["id"], featureInstanceId=updated.get("featureInstanceId") or self.featureInstanceId or "", name=updated.get("name"), status=updated.get("status", "running"), currentRound=updated.get("currentRound", 0) or 0, lastActivity=updated.get("lastActivity", getUtcTimestamp()), startedAt=updated.get("startedAt", getUtcTimestamp()), workflowMode=ChatbotWorkflowModeEnum(updated.get("workflowMode", "Chatbot")), maxSteps=max_steps if max_steps is not None else 10, logs=[], messages=[], ) def createConversation(self, conversationData: Dict[str, Any]) -> ChatbotConversation: """Creates a new conversation if user has permission.""" if not self.checkRbacPermission(ChatbotConversation, "create"): raise PermissionError("No permission to create conversations") # Set timestamp if not present currentTime = getUtcTimestamp() if "startedAt" not in conversationData: conversationData["startedAt"] = currentTime if "lastActivity" not in conversationData: conversationData["lastActivity"] = currentTime # Set featureInstanceId from context (no mandateId in DB) if "featureInstanceId" not in conversationData or not conversationData["featureInstanceId"]: conversationData["featureInstanceId"] = self.featureInstanceId or "" if not conversationData.get("featureInstanceId"): conversationData["featureInstanceId"] = self.featureInstanceId or "" # Use generic field separation - logs/messages go to objectFields, not stored simpleFields, objectFields = self._separateObjectFields(ChatbotConversation, conversationData) # Create conversation in database created = self.db.recordCreate(ChatbotConversation, simpleFields) return ChatbotConversation( id=created["id"], featureInstanceId=created.get("featureInstanceId") or self.featureInstanceId or "", name=created.get("name"), status=created.get("status", "running"), currentRound=created.get("currentRound", 0) or 0, lastActivity=created.get("lastActivity", currentTime), startedAt=created.get("startedAt", currentTime), workflowMode=ChatbotWorkflowModeEnum(created.get("workflowMode", "Chatbot")), maxSteps=created.get("maxSteps", 10), logs=[], messages=[] ) def createWorkflow(self, workflowData: Dict[str, Any]) -> ChatbotConversation: """Backward-compat alias: maps workflowData to conversationData.""" return self.createConversation(workflowData) def updateConversation(self, conversationId: str, conversationData: Dict[str, Any]) -> ChatbotConversation: """Updates a conversation if user has access.""" conv = self.getConversation(conversationId) if not conv: return None if not self.checkRbacPermission(ChatbotConversation, "update", conversationId): raise PermissionError(f"No permission to update conversation {conversationId}") simpleFields, objectFields = self._separateObjectFields(ChatbotConversation, conversationData) simpleFields["lastActivity"] = getUtcTimestamp() updated = self.db.recordModify(ChatbotConversation, conversationId, simpleFields) # Reuse logs/messages from conv — update only touches simple fields, not related data return ChatbotConversation( id=updated["id"], featureInstanceId=updated.get("featureInstanceId") or conv.featureInstanceId or self.featureInstanceId or "", name=updated.get("name", conv.name), status=updated.get("status", conv.status), currentRound=updated.get("currentRound", conv.currentRound), lastActivity=updated.get("lastActivity", conv.lastActivity), startedAt=updated.get("startedAt", conv.startedAt), workflowMode=ChatbotWorkflowModeEnum(updated.get("workflowMode", conv.workflowMode.value)), maxSteps=updated.get("maxSteps") if updated.get("maxSteps") is not None else conv.maxSteps, logs=conv.logs, messages=conv.messages ) def updateWorkflow(self, workflowId: str, workflowData: Dict[str, Any]) -> ChatbotConversation: """Backward-compat alias.""" return self.updateConversation(workflowId, workflowData) def deleteConversation(self, conversationId: str) -> bool: """Deletes a conversation and all related data if user has access.""" try: conv = self.getConversation(conversationId) if not conv: return False if not self.checkRbacPermission(ChatbotConversation, "delete", conversationId): raise PermissionError(f"No permission to delete conversation {conversationId}") # CASCADE DELETE: Delete all related data first # 1. Delete all messages and their documents messages = self.getMessages(conversationId) for message in messages: messageId = message.id if messageId: existing_docs = getRecordsetWithRBAC(self.db, ChatbotDocument, self.currentUser, recordFilter={"messageId": messageId}, featureCode="chatbot") for doc in existing_docs: self.db.recordDelete(ChatbotDocument, doc["id"]) self.db.recordDelete(ChatbotMessage, messageId) # 2. Delete conversation logs existing_logs = getRecordsetWithRBAC(self.db, ChatbotLog, self.currentUser, recordFilter={"conversationId": conversationId}, featureCode="chatbot") for log in existing_logs: self.db.recordDelete(ChatbotLog, log["id"]) # 3. Delete the conversation success = self.db.recordDelete(ChatbotConversation, conversationId) return success except Exception as e: logger.error(f"Error deleting conversation {conversationId}: {str(e)}") return False def deleteWorkflow(self, workflowId: str) -> bool: """Backward-compat alias.""" return self.deleteConversation(workflowId) # Message methods def getMessages(self, conversationId: str, pagination: Optional[PaginationParams] = None) -> Union[List[ChatbotMessage], PaginatedResult]: """ Returns messages for a conversation if user has access. Supports optional pagination, sorting, and filtering. Args: conversationId: The conversation ID (workflowId for backward compat) pagination: Optional pagination parameters. If None, returns all items. Returns: If pagination is None: List[ChatbotMessage] If pagination is provided: PaginatedResult with items and metadata """ # Check conversation access first conversations = getRecordsetWithRBAC(self.db, ChatbotConversation, self.currentUser, recordFilter={"id": conversationId}, mandateId=self.mandateId, featureInstanceId=self.featureInstanceId, featureCode="chatbot" ) if not conversations: if pagination is None: return [] return PaginatedResult(items=[], totalItems=0, totalPages=0) # Get messages for this conversation messages = getRecordsetWithRBAC(self.db, ChatbotMessage, self.currentUser, recordFilter={"conversationId": conversationId}, mandateId=self.mandateId, featureInstanceId=self.featureInstanceId, featureCode="chatbot") # Convert raw messages to dict format for sorting/filtering messageDicts = [] for msg in messages: messageDicts.append({ "id": msg.get("id"), "conversationId": msg.get("conversationId"), "parentMessageId": msg.get("parentMessageId"), "documentsLabel": msg.get("documentsLabel"), "message": msg.get("message"), "role": msg.get("role", "assistant"), "status": msg.get("status", "step"), "sequenceNr": msg.get("sequenceNr", 0), "publishedAt": msg.get("publishedAt", msg.get("timestamp", getUtcTimestamp())), "success": msg.get("success"), "actionId": msg.get("actionId"), "actionMethod": msg.get("actionMethod"), "actionName": msg.get("actionName"), "roundNumber": msg.get("roundNumber"), "taskNumber": msg.get("taskNumber"), "actionNumber": msg.get("actionNumber"), "taskProgress": msg.get("taskProgress"), "actionProgress": msg.get("actionProgress") }) # Apply default sorting by publishedAt if no sort specified if pagination is None or not pagination.sort: messageDicts.sort(key=lambda x: x.get("publishedAt", getUtcTimestamp())) # Apply filtering (if filters provided) if pagination and pagination.filters: messageDicts = self._applyFilters(messageDicts, pagination.filters) # Apply sorting (in order of sortFields) if pagination and pagination.sort: messageDicts = self._applySorting(messageDicts, pagination.sort) # If no pagination requested, return all items (batch-fetch documents to avoid N+1) if pagination is None: msg_ids = [m["id"] for m in messageDicts] docs_by_message = self.getDocumentsForMessages(msg_ids) if msg_ids else {} chat_messages = [] for msg in messageDicts: documents = docs_by_message.get(msg["id"], []) chat_message = ChatbotMessage( id=msg["id"], conversationId=msg["conversationId"], parentMessageId=msg.get("parentMessageId"), documents=documents, documentsLabel=msg.get("documentsLabel"), message=msg.get("message"), role=msg.get("role", "assistant"), status=msg.get("status", "step"), sequenceNr=msg.get("sequenceNr", 0), publishedAt=msg.get("publishedAt", getUtcTimestamp()), success=msg.get("success"), actionId=msg.get("actionId"), actionMethod=msg.get("actionMethod"), actionName=msg.get("actionName"), roundNumber=msg.get("roundNumber"), taskNumber=msg.get("taskNumber"), actionNumber=msg.get("actionNumber"), taskProgress=msg.get("taskProgress"), actionProgress=msg.get("actionProgress") ) chat_messages.append(chat_message) return chat_messages # Count total items after filters totalItems = len(messageDicts) totalPages = math.ceil(totalItems / pagination.pageSize) if totalItems > 0 else 0 # Apply pagination (skip/limit) startIdx = (pagination.page - 1) * pagination.pageSize endIdx = startIdx + pagination.pageSize pagedMessageDicts = messageDicts[startIdx:endIdx] paged_msg_ids = [m["id"] for m in pagedMessageDicts] docs_by_message = self.getDocumentsForMessages(paged_msg_ids) if paged_msg_ids else {} chat_messages = [] for msg in pagedMessageDicts: documents = docs_by_message.get(msg["id"], []) chat_message = ChatbotMessage( id=msg["id"], conversationId=msg["conversationId"], parentMessageId=msg.get("parentMessageId"), documents=documents, documentsLabel=msg.get("documentsLabel"), message=msg.get("message"), role=msg.get("role", "assistant"), status=msg.get("status", "step"), sequenceNr=msg.get("sequenceNr", 0), publishedAt=msg.get("publishedAt", getUtcTimestamp()), success=msg.get("success"), actionId=msg.get("actionId"), actionMethod=msg.get("actionMethod"), actionName=msg.get("actionName"), roundNumber=msg.get("roundNumber"), taskNumber=msg.get("taskNumber"), actionNumber=msg.get("actionNumber"), taskProgress=msg.get("taskProgress"), actionProgress=msg.get("actionProgress") ) chat_messages.append(chat_message) return PaginatedResult( items=chat_messages, totalItems=totalItems, totalPages=totalPages ) def createMessage(self, messageData: Dict[str, Any], event_manager=None) -> ChatbotMessage: """Creates a message for a conversation if user has access. Accepts workflowId (from bridge) or conversationId.""" try: if "id" not in messageData or not messageData["id"]: messageData["id"] = f"msg_{uuid.uuid4()}" # Map workflowId to conversationId (bridge compatibility) if "workflowId" in messageData and "conversationId" not in messageData: messageData["conversationId"] = messageData["workflowId"] requiredFields = ["id", "conversationId"] for field in requiredFields: if field not in messageData: logger.error(f"Required field '{field}' missing in messageData") raise ValueError(f"Required field '{field}' missing in message data") conversationId = messageData["conversationId"] conv = self.getConversation(conversationId) if not conv: raise PermissionError(f"No access to conversation {conversationId}") if not self.checkRbacPermission(ChatbotConversation, "update", conversationId): raise PermissionError(f"No permission to modify conversation {conversationId}") # Validate that ID is not None if messageData["id"] is None: messageData["id"] = f"msg_{uuid.uuid4()}" logger.warning(f"Automatically generated ID for workflow message: {messageData['id']}") # Set status if not present if "status" not in messageData: messageData["status"] = "step" # Default status for intermediate messages # Ensure role and agentName are present if "role" not in messageData: messageData["role"] = "assistant" if messageData.get("agentName") else "user" if "agentName" not in messageData: messageData["agentName"] = "" # Set roundNumber, taskNumber, actionNumber if not provided if "roundNumber" not in messageData: messageData["roundNumber"] = conv.currentRound if "taskNumber" not in messageData: messageData["taskNumber"] = 0 if "actionNumber" not in messageData: messageData["actionNumber"] = 0 # Use generic field separation - no mandateId/featureInstanceId in message simpleFields, objectFields = self._separateObjectFields(ChatbotMessage, messageData) # Handle documents separately - they will be stored in normalized documents table documents_to_create = objectFields.get("documents", []) createdMessage = self.db.recordCreate(ChatbotMessage, simpleFields) created_documents = [] logger.debug(f"Creating {len(documents_to_create)} document(s) for message {createdMessage['id']}") for idx, doc_data in enumerate(documents_to_create): try: if isinstance(doc_data, ChatbotDocument): doc_dict = doc_data.model_dump() elif isinstance(doc_data, dict): doc_dict = dict(doc_data) else: try: doc_dict = ChatbotDocument(**doc_data).model_dump() except Exception as e: logger.error(f"Invalid document data type for message creation (document {idx + 1}/{len(documents_to_create)}): {e}") continue # Ensure messageId is set doc_dict["messageId"] = createdMessage["id"] logger.debug(f"Creating document {idx + 1}/{len(documents_to_create)}: fileName={doc_dict.get('fileName', 'unknown')}, fileId={doc_dict.get('fileId', 'unknown')}, messageId={doc_dict.get('messageId', 'unknown')}") created_doc = self.createDocument(doc_dict) if created_doc: created_documents.append(created_doc) logger.debug(f"Successfully created document {idx + 1}/{len(documents_to_create)}: {created_doc.fileName} (id: {created_doc.id})") else: logger.error(f"Failed to create document {idx + 1}/{len(documents_to_create)}: createDocument returned None for fileName={doc_dict.get('fileName', 'unknown')}") except Exception as e: logger.error(f"Error processing document {idx + 1}/{len(documents_to_create)}: {e}", exc_info=True) logger.info(f"Created {len(created_documents)}/{len(documents_to_create)} document(s) for message {createdMessage['id']}") chat_message = ChatbotMessage( id=createdMessage["id"], conversationId=createdMessage["conversationId"], parentMessageId=createdMessage.get("parentMessageId"), documents=created_documents, documentsLabel=createdMessage.get("documentsLabel"), message=createdMessage.get("message"), role=createdMessage.get("role", "assistant"), status=createdMessage.get("status", "step"), sequenceNr=len(conv.messages) + 1, publishedAt=createdMessage.get("publishedAt", getUtcTimestamp()), roundNumber=createdMessage.get("roundNumber"), taskNumber=createdMessage.get("taskNumber"), actionNumber=createdMessage.get("actionNumber"), success=createdMessage.get("success"), actionId=createdMessage.get("actionId"), actionMethod=createdMessage.get("actionMethod"), actionName=createdMessage.get("actionName") ) if event_manager: try: message_timestamp = parseTimestamp(chat_message.publishedAt, default=getUtcTimestamp()) asyncio.create_task(event_manager.emit_event( context_id=conversationId, event_type="chatdata", data={ "type": "message", "createdAt": message_timestamp, "item": chat_message.model_dump() }, event_category="chat" )) except Exception as e: logger.debug(f"Could not emit message event: {e}") # Debug: Store message and documents for debugging - only if debug enabled storeDebugMessageAndDocuments(chat_message, self.currentUser, mandateId=self.mandateId, featureInstanceId=self.featureInstanceId) return chat_message except Exception as e: logger.error(f"Error creating workflow message: {str(e)}") return None def updateMessage(self, messageId: str, messageData: Dict[str, Any]) -> Optional[ChatbotMessage]: """Updates a conversation message if user has access.""" try: if not messageId: logger.error("No messageId provided for updateMessage") raise ValueError("messageId cannot be empty") if "workflowId" in messageData and "conversationId" not in messageData: messageData["conversationId"] = messageData["workflowId"] messages = getRecordsetWithRBAC(self.db, ChatbotMessage, self.currentUser, recordFilter={"id": messageId}, mandateId=self.mandateId, featureInstanceId=self.featureInstanceId, featureCode="chatbot") if not messages: logger.warning(f"Message with ID {messageId} does not exist in database") if "conversationId" in messageData: conv = self.getConversation(messageData["conversationId"]) if not conv: raise PermissionError(f"No access to conversation {messageData['conversationId']}") if not self.checkRbacPermission(ChatbotConversation, "update", messageData["conversationId"]): raise PermissionError(f"No permission to modify conversation") logger.info(f"Creating new message with ID {messageId} for conversation {messageData['conversationId']}") created = self.db.recordCreate(ChatbotMessage, messageData) return ChatbotMessage(**created) if created else None logger.error("Conversation ID missing for new message") return None existingMessage = messages[0] conversationId = existingMessage.get("conversationId") conv = self.getConversation(conversationId) if not conv: raise PermissionError(f"No access to conversation {conversationId}") if not self.checkRbacPermission(ChatbotConversation, "update", conversationId): raise PermissionError(f"No permission to modify conversation {conversationId}") simpleFields, objectFields = self._separateObjectFields(ChatbotMessage, messageData) if "role" not in simpleFields and "role" not in existingMessage: simpleFields["role"] = "assistant" if 'id' not in simpleFields: simpleFields['id'] = messageId if "createdAt" in simpleFields and "startedAt" not in simpleFields: simpleFields["startedAt"] = simpleFields["createdAt"] del simpleFields["createdAt"] updatedMessage = self.db.recordModify(ChatbotMessage, messageId, simpleFields) if 'documents' in objectFields: for doc_data in objectFields['documents']: try: if isinstance(doc_data, ChatbotDocument): doc_dict = doc_data.model_dump() elif isinstance(doc_data, dict): doc_dict = dict(doc_data) else: doc_dict = ChatbotDocument(**doc_data).model_dump() doc_dict["messageId"] = messageId self.createDocument(doc_dict) except Exception as e: logger.error(f"Error updating message documents: {e}") if not updatedMessage: logger.warning(f"Failed to update message {messageId}") return None return ChatbotMessage(**updatedMessage) except Exception as e: logger.error(f"Error updating message {messageId}: {str(e)}", exc_info=True) raise ValueError(f"Error updating message {messageId}: {str(e)}") def deleteMessage(self, conversationId: str, messageId: str) -> bool: """Deletes a conversation message and related data if user has access.""" try: conv = self.getConversation(conversationId) if not conv: logger.warning(f"No access to conversation {conversationId}") return False if not self.checkRbacPermission(ChatbotConversation, "update", conversationId): raise PermissionError(f"No permission to modify conversation {conversationId}") messages = self.getMessages(conversationId) message = next((m for m in messages if m.id == messageId), None) if not message: logger.warning(f"Message {messageId} for conversation {conversationId} not found") return False existing_docs = getRecordsetWithRBAC(self.db, ChatbotDocument, self.currentUser, recordFilter={"messageId": messageId}, featureCode="chatbot") for doc in existing_docs: self.db.recordDelete(ChatbotDocument, doc["id"]) success = self.db.recordDelete(ChatbotMessage, messageId) return success except Exception as e: logger.error(f"Error deleting message {messageId}: {str(e)}") return False def deleteFileFromMessage(self, conversationId: str, messageId: str, fileId: str) -> bool: """Removes a file reference from a message if user has access.""" try: conv = self.getConversation(conversationId) if not conv: logger.warning(f"No access to conversation {conversationId}") return False if not self.checkRbacPermission(ChatbotConversation, "update", conversationId): raise PermissionError(f"No permission to modify conversation {conversationId}") documents = getRecordsetWithRBAC(self.db, ChatbotDocument, self.currentUser, recordFilter={"messageId": messageId}, featureCode="chatbot") if not documents: logger.warning(f"No documents found for message {messageId}") return False # Find and delete the specific document removed = False for doc in documents: docId = doc.get("id") fileIdValue = doc.get("fileId") # Flexible matching approach shouldRemove = ( (docId == fileId) or (fileIdValue == fileId) or (isinstance(docId, str) and str(fileId) in docId) or (isinstance(fileIdValue, str) and str(fileId) in fileIdValue) ) if shouldRemove: # Delete the document from normalized table success = self.db.recordDelete(ChatbotDocument, docId) if success: removed = True else: logger.warning(f"Failed to delete document {docId}") if not removed: logger.warning(f"No matching file {fileId} found in message {messageId}") return False except Exception as e: logger.error(f"Error removing file {fileId} from message {messageId}: {str(e)}") return False # Document methods def getDocuments(self, messageId: str) -> List[ChatbotDocument]: """Returns documents for a message from normalized table.""" try: documents = getRecordsetWithRBAC(self.db, ChatbotDocument, self.currentUser, recordFilter={"messageId": messageId}, mandateId=self.mandateId, featureInstanceId=self.featureInstanceId, featureCode="chatbot") return [ChatbotDocument(**doc) for doc in documents] except Exception as e: logger.error(f"Error getting message documents: {str(e)}") return [] def getDocumentsForMessages(self, messageIds: List[str]) -> Dict[str, List[ChatbotDocument]]: """Returns documents for multiple messages in one query. Returns {messageId: [doc, ...]}.""" if not messageIds: return {} try: documents = getRecordsetWithRBAC( self.db, ChatbotDocument, self.currentUser, recordFilter={"messageId": messageIds}, mandateId=self.mandateId, featureInstanceId=self.featureInstanceId, featureCode="chatbot", ) result: Dict[str, List[ChatbotDocument]] = {mid: [] for mid in messageIds} for doc in documents: mid = doc.get("messageId") if mid in result: result[mid].append(ChatbotDocument(**doc)) return result except Exception as e: logger.error(f"Error getting documents for messages: {e}") return {mid: [] for mid in messageIds} def createDocument(self, documentData: Dict[str, Any]) -> ChatbotDocument: """Creates a document for a message in normalized table.""" try: document = ChatbotDocument(**documentData) logger.debug(f"Creating document: fileName={document.fileName}, fileId={document.fileId}, messageId={document.messageId}") created = self.db.recordCreate(ChatbotDocument, document.model_dump()) if created: return ChatbotDocument(**created) logger.error(f"Failed to create document for fileName={document.fileName}") return None except Exception as e: logger.error(f"Error creating message document: {str(e)}", exc_info=True) return None # Log methods def getLogs(self, conversationId: str, pagination: Optional[PaginationParams] = None) -> Union[List[ChatbotLog], PaginatedResult]: """ Returns logs for a conversation if user has access. Supports optional pagination, sorting, and filtering. """ conversations = getRecordsetWithRBAC(self.db, ChatbotConversation, self.currentUser, recordFilter={"id": conversationId}, mandateId=self.mandateId, featureInstanceId=self.featureInstanceId, featureCode="chatbot" ) if not conversations: if pagination is None: return [] return PaginatedResult(items=[], totalItems=0, totalPages=0) logs = getRecordsetWithRBAC(self.db, ChatbotLog, self.currentUser, recordFilter={"conversationId": conversationId}, mandateId=self.mandateId, featureInstanceId=self.featureInstanceId, featureCode="chatbot") logDicts = [] for log in logs: logDicts.append({ "id": log.get("id"), "conversationId": log.get("conversationId"), "message": log.get("message"), "type": log.get("type"), "timestamp": log.get("timestamp", getUtcTimestamp()), "status": log.get("status"), "progress": log.get("progress") }) # Apply default sorting by timestamp if no sort specified if pagination is None or not pagination.sort: logDicts.sort(key=lambda x: parseTimestamp(x.get("timestamp"), default=0)) # Apply filtering (if filters provided) if pagination and pagination.filters: logDicts = self._applyFilters(logDicts, pagination.filters) # Apply sorting (in order of sortFields) if pagination and pagination.sort: logDicts = self._applySorting(logDicts, pagination.sort) if pagination is None: return [ChatbotLog(**log) for log in logDicts] totalItems = len(logDicts) totalPages = math.ceil(totalItems / pagination.pageSize) if totalItems > 0 else 0 startIdx = (pagination.page - 1) * pagination.pageSize endIdx = startIdx + pagination.pageSize pagedLogDicts = logDicts[startIdx:endIdx] items = [ChatbotLog(**log) for log in pagedLogDicts] return PaginatedResult( items=items, totalItems=totalItems, totalPages=totalPages ) def createLog(self, logData: Dict[str, Any], event_manager=None) -> Optional[ChatbotLog]: """Creates a log entry for a conversation if user has access. Accepts workflowId for backward compat.""" conversationId = logData.get("conversationId") or logData.get("workflowId") if not conversationId: logger.error("No conversationId/workflowId provided for createLog") return None conv = self.getConversation(conversationId) if not conv: logger.warning(f"No access to conversation {conversationId}") return None if not self.checkRbacPermission(ChatbotConversation, "update", conversationId): logger.warning(f"No permission to modify conversation {conversationId}") return None if "timestamp" not in logData: logData["timestamp"] = getUtcTimestamp() logData["conversationId"] = conversationId if "status" not in logData and "type" in logData: logData["status"] = "error" if logData["type"] == "error" else "running" if "progress" not in logData: logData["progress"] = 1.0 if logData.get("type") == "error" else 0.5 try: log_model = ChatbotLog(**logData) except Exception as e: logger.error(f"Invalid log data: {e}") return None createdLog = self.db.recordCreate(ChatbotLog, log_model.model_dump()) if not createdLog: return None if event_manager: try: log_timestamp = parseTimestamp(createdLog.get("timestamp"), default=getUtcTimestamp()) asyncio.create_task(event_manager.emit_event( context_id=conversationId, event_type="chatdata", data={"type": "log", "createdAt": log_timestamp, "item": ChatbotLog(**createdLog).model_dump()}, event_category="log", message="New log" )) except Exception as e: logger.debug(f"Could not emit log event: {e}") return ChatbotLog(**createdLog) def getUnifiedChatData(self, conversationId: str, afterTimestamp: Optional[float] = None) -> Dict[str, Any]: """ Returns unified chat data (messages, logs) for a conversation in chronological order. Uses timestamp-based selective data transfer for efficient polling. """ conversations = getRecordsetWithRBAC(self.db, ChatbotConversation, self.currentUser, recordFilter={"id": conversationId}, mandateId=self.mandateId, featureInstanceId=self.featureInstanceId, featureCode="chatbot" ) if not conversations: return {"items": []} items = [] messages = getRecordsetWithRBAC(self.db, ChatbotMessage, self.currentUser, recordFilter={"conversationId": conversationId}, mandateId=self.mandateId, featureInstanceId=self.featureInstanceId, featureCode="chatbot") # Batch-fetch documents for all messages (avoids N+1) message_ids = [m["id"] for m in messages if afterTimestamp is None or parseTimestamp(m.get("publishedAt"), default=getUtcTimestamp()) > afterTimestamp] docs_by_message = self.getDocumentsForMessages(message_ids) if message_ids else {} for msg in messages: msgTimestamp = parseTimestamp(msg.get("publishedAt"), default=getUtcTimestamp()) if afterTimestamp is not None and msgTimestamp <= afterTimestamp: continue documents = docs_by_message.get(msg["id"], []) chatMessage = ChatbotMessage( id=msg["id"], conversationId=msg["conversationId"], parentMessageId=msg.get("parentMessageId"), documents=documents, documentsLabel=msg.get("documentsLabel"), message=msg.get("message"), role=msg.get("role", "assistant"), status=msg.get("status", "step"), sequenceNr=msg.get("sequenceNr", 0), publishedAt=msg.get("publishedAt", getUtcTimestamp()), success=msg.get("success"), actionId=msg.get("actionId"), actionMethod=msg.get("actionMethod"), actionName=msg.get("actionName"), roundNumber=msg.get("roundNumber"), taskNumber=msg.get("taskNumber"), actionNumber=msg.get("actionNumber"), taskProgress=msg.get("taskProgress"), actionProgress=msg.get("actionProgress") ) items.append({"type": "message", "createdAt": msgTimestamp, "item": chatMessage}) logs = getRecordsetWithRBAC(self.db, ChatbotLog, self.currentUser, recordFilter={"conversationId": conversationId}, mandateId=self.mandateId, featureInstanceId=self.featureInstanceId, featureCode="chatbot") for log in logs: logTimestamp = parseTimestamp(log.get("timestamp"), default=getUtcTimestamp()) if afterTimestamp is not None and logTimestamp <= afterTimestamp: continue chatLog = ChatbotLog(**log) items.append({"type": "log", "createdAt": logTimestamp, "item": chatLog}) items.sort(key=lambda x: parseTimestamp(x.get("createdAt"), default=0)) return {"items": items} def getInterface(currentUser: Optional[User] = None, mandateId: Optional[str] = None, featureInstanceId: Optional[str] = None) -> 'ChatObjects': """ Returns a ChatObjects instance for the current user. Handles initialization of database and records. Args: currentUser: The authenticated user mandateId: The mandate ID from RequestContext (X-Mandate-Id header). Required. featureInstanceId: The feature instance ID from RequestContext (X-Feature-Instance-Id header). """ if not currentUser: raise ValueError("Invalid user context: user is required") effectiveMandateId = str(mandateId) if mandateId else None effectiveFeatureInstanceId = str(featureInstanceId) if featureInstanceId else None # Create context key including featureInstanceId for proper isolation contextKey = f"{effectiveMandateId}_{effectiveFeatureInstanceId}_{currentUser.id}" # Create new instance if not exists if contextKey not in _chatInterfaces: _chatInterfaces[contextKey] = ChatObjects(currentUser, mandateId=effectiveMandateId, featureInstanceId=effectiveFeatureInstanceId) else: # Update user context if needed _chatInterfaces[contextKey].setUserContext(currentUser, mandateId=effectiveMandateId, featureInstanceId=effectiveFeatureInstanceId) return _chatInterfaces[contextKey]