""" Interface to LucyDOM database and AI Connectors. Uses the JSON connector for data access with added language support. """ import os import logging import uuid from datetime import datetime, UTC, timezone from typing import Dict, Any, List, Optional, Union import asyncio from modules.interfaces.interfaceChatAccess import ChatAccess from modules.interfaces.interfaceChatModel import ( TaskStatus, UserInputRequest, ChatDocument, TaskItem, ChatStat, ChatLog, ChatMessage, ChatWorkflow, TaskAction, TaskResult, ActionResult ) from modules.interfaces.interfaceAppModel import User # DYNAMIC PART: Connectors to the Interface from modules.connectors.connectorDbJson import DatabaseConnector from modules.shared.timezoneUtils import get_utc_timestamp # Basic Configurations from modules.shared.configuration import APP_CONFIG logger = logging.getLogger(__name__) # Singleton factory for Chat instances _chatInterfaces = {} 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): """Initializes the Chat Interface.""" # Initialize variables self.currentUser = currentUser # Store User object directly self.userId = currentUser.id if currentUser else None self.mandateId = currentUser.mandateId if currentUser else None self.access = None # Will be set when user context is provided # Initialize services self._initializeServices() # Initialize database self._initializeDatabase() # Set user context if provided if currentUser: self.setUserContext(currentUser) def _initializeServices(self): pass def setUserContext(self, currentUser: User): """Sets the user context for the interface.""" self.currentUser = currentUser # Store User object directly self.userId = currentUser.id self.mandateId = currentUser.mandateId if not self.userId or not self.mandateId: raise ValueError("Invalid user context: id and mandateId are required") # Add language settings self.userLanguage = currentUser.language # Default user language # Initialize access control with user context self.access = ChatAccess(self.currentUser, self.db) # Convert to dict only when needed # Update database context self.db.updateContext(self.userId) logger.debug(f"User context set: userId={self.userId}, mandateId={self.mandateId}") def _initializeDatabase(self): """Initializes the database connection.""" try: # Get configuration values with defaults dbHost = APP_CONFIG.get("DB_CHAT_HOST", "_no_config_default_data") dbDatabase = APP_CONFIG.get("DB_CHAT_DATABASE", "chat") dbUser = APP_CONFIG.get("DB_CHAT_USER") dbPassword = APP_CONFIG.get("DB_CHAT_PASSWORD_SECRET") # Ensure the database directory exists os.makedirs(dbHost, exist_ok=True) self.db = DatabaseConnector( dbHost=dbHost, dbDatabase=dbDatabase, dbUser=dbUser, dbPassword=dbPassword ) 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 _uam(self, table: str, recordset: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """Delegate to access control module.""" # First apply access control filteredRecords = self.access.uam(table, recordset) # Then filter out database-specific fields cleanedRecords = [] for record in filteredRecords: # Create a new dict with only non-database fields cleanedRecord = {k: v for k, v in record.items() if not k.startswith('_')} cleanedRecords.append(cleanedRecord) return cleanedRecords def _canModify(self, table: str, recordId: Optional[str] = None) -> bool: """Delegate to access control module.""" return self.access.canModify(table, recordId) def _clearTableCache(self, table: str) -> None: """Clears the cache for a specific table to ensure fresh data.""" self.db.clearTableCache(table) # Utilities def getInitialId(self, table: str) -> Optional[str]: """Returns the initial ID for a table.""" return self.db.getInitialId(table) # Workflow methods def getAllWorkflows(self) -> List[Dict[str, Any]]: """Returns workflows based on user access level.""" allWorkflows = self.db.getRecordset("workflows") return self._uam("workflows", allWorkflows) def getWorkflow(self, workflowId: str) -> Optional[ChatWorkflow]: """Returns a workflow by ID if user has access.""" workflows = self.db.getRecordset("workflows", recordFilter={"id": workflowId}) if not workflows: return None filteredWorkflows = self._uam("workflows", workflows) if not filteredWorkflows: return None workflow = filteredWorkflows[0] try: # Validate workflow data against ChatWorkflow model return ChatWorkflow( id=workflow["id"], status=workflow.get("status", "running"), name=workflow.get("name"), currentRound=workflow.get("currentRound", 0), # Default value currentTask=workflow.get("currentTask", 0), currentAction=workflow.get("currentAction", 0), totalTasks=workflow.get("totalTasks", 0), totalActions=workflow.get("totalActions", 0), lastActivity=workflow.get("lastActivity", get_utc_timestamp()), startedAt=workflow.get("startedAt", get_utc_timestamp()), logs=[ChatLog(**log) for log in workflow.get("logs", [])], messages=[ChatMessage(**msg) for msg in workflow.get("messages", [])], stats=ChatStat(**workflow.get("stats", {})) if workflow.get("stats") else None, mandateId=workflow.get("mandateId", self.currentUser.mandateId) ) except Exception as e: logger.error(f"Error validating workflow data: {str(e)}") return None def createWorkflow(self, workflowData: Dict[str, Any]) -> ChatWorkflow: """Creates a new workflow if user has permission.""" if not self._canModify("workflows"): raise PermissionError("No permission to create workflows") # Set timestamp if not present currentTime = get_utc_timestamp() if "startedAt" not in workflowData: workflowData["startedAt"] = currentTime if "lastActivity" not in workflowData: workflowData["lastActivity"] = currentTime # Create workflow in database created = self.db.recordCreate("workflows", workflowData) # Clear cache to ensure fresh data self._clearTableCache("workflows") # Convert to ChatWorkflow model return ChatWorkflow( id=created["id"], status=created.get("status", "running"), name=created.get("name"), currentRound=created.get("currentRound", 0), # Default value currentTask=created.get("currentTask", 0), currentAction=created.get("currentAction", 0), totalTasks=created.get("totalTasks", 0), totalActions=created.get("totalActions", 0), lastActivity=created.get("lastActivity", currentTime), startedAt=created.get("startedAt", currentTime), logs=[], messages=[], stats=ChatStat(**created.get("stats", {})) if created.get("stats") else None, mandateId=created.get("mandateId", self.currentUser.mandateId) ) def updateWorkflow(self, workflowId: str, workflowData: Dict[str, Any]) -> ChatWorkflow: """Updates a workflow if user has access.""" # Check if the workflow exists and user has access workflow = self.getWorkflow(workflowId) if not workflow: return None if not self._canModify("workflows", workflowId): raise PermissionError(f"No permission to update workflow {workflowId}") # Set update time workflowData["lastActivity"] = get_utc_timestamp() # Update workflow in database updated = self.db.recordModify("workflows", workflowId, workflowData) # Clear cache to ensure fresh data self._clearTableCache("workflows") # Convert to ChatWorkflow model return ChatWorkflow( id=updated["id"], status=updated.get("status", workflow.status), name=updated.get("name", workflow.name), currentRound=updated.get("currentRound", workflow.currentRound), currentTask=updated.get("currentTask", workflow.currentTask), currentAction=updated.get("currentAction", workflow.currentAction), totalTasks=updated.get("totalTasks", workflow.totalTasks), totalActions=updated.get("totalActions", workflow.totalActions), lastActivity=updated.get("lastActivity", workflow.lastActivity), startedAt=updated.get("startedAt", workflow.startedAt), logs=[ChatLog(**log) for log in updated.get("logs", workflow.logs)], messages=[ChatMessage(**msg) for msg in updated.get("messages", workflow.messages)], stats=ChatStat(**updated.get("stats", workflow.stats.dict() if workflow.stats else {})) if updated.get("stats") or workflow.stats else None, mandateId=updated.get("mandateId", workflow.mandateId) ) def deleteWorkflow(self, workflowId: str) -> bool: """Deletes a workflow if user has access.""" # Check if the workflow exists and user has access workflow = self.getWorkflow(workflowId) if not workflow: return False if not self._canModify("workflows", workflowId): raise PermissionError(f"No permission to delete workflow {workflowId}") # Delete workflow success = self.db.recordDelete("workflows", workflowId) # Clear cache to ensure fresh data self._clearTableCache("workflows") return success # Workflow Messages def getWorkflowMessages(self, workflowId: str) -> List[ChatMessage]: """Returns messages for a workflow if user has access to the workflow.""" # Check workflow access first workflow = self.getWorkflow(workflowId) if not workflow: return [] # Get messages for this workflow messages = self.db.getRecordset("workflowMessages", recordFilter={"workflowId": workflowId}) # Sort messages by publishedAt timestamp to ensure chronological order messages.sort(key=lambda x: x.get("publishedAt", x.get("timestamp", "0"))) # Convert messages to ChatMessage objects with proper document handling chat_messages = [] for msg in messages: # Ensure documents field is properly converted to ChatDocument objects if "documents" in msg and msg["documents"]: try: # Convert each document back to ChatDocument object documents = [] for doc in msg["documents"]: if isinstance(doc, dict): documents.append(ChatDocument(**doc)) else: documents.append(doc) msg["documents"] = documents except Exception as e: logger.warning(f"Error converting documents for message {msg.get('id', 'unknown')}: {e}") msg["documents"] = [] else: msg["documents"] = [] chat_messages.append(ChatMessage(**msg)) return chat_messages def createWorkflowMessage(self, messageData: Dict[str, Any]) -> ChatMessage: """Creates a message for a workflow if user has access.""" try: # Ensure ID is present if "id" not in messageData or not messageData["id"]: messageData["id"] = f"msg_{uuid.uuid4()}" # Check required fields requiredFields = ["id", "workflowId"] 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") # Check workflow access workflowId = messageData["workflowId"] workflow = self.getWorkflow(workflowId) if not workflow: raise PermissionError(f"No access to workflow {workflowId}") if not self._canModify("workflows", workflowId): raise PermissionError(f"No permission to modify workflow {workflowId}") # 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"] = "" # CRITICAL FIX: Automatically set roundNumber, taskNumber, and actionNumber if not provided # This ensures messages have the correct progress context when workflows are continued if "roundNumber" not in messageData: messageData["roundNumber"] = workflow.currentRound logger.debug(f"Auto-setting roundNumber to {workflow.currentRound} for message {messageData['id']}") if "taskNumber" not in messageData: messageData["taskNumber"] = workflow.currentTask logger.debug(f"Auto-setting taskNumber to {workflow.currentTask} for message {messageData['id']}") if "actionNumber" not in messageData: messageData["actionNumber"] = workflow.currentAction logger.debug(f"Auto-setting actionNumber to {workflow.currentAction} for message {messageData['id']}") # Convert ChatDocument objects to dictionaries for database storage if "documents" in messageData and messageData["documents"]: documents_for_db = [] for doc in messageData["documents"]: if isinstance(doc, ChatDocument): # Convert ChatDocument to dictionary documents_for_db.append(doc.dict()) else: # Already a dictionary documents_for_db.append(doc) messageData["documents"] = documents_for_db # Create message in database createdMessage = self.db.recordCreate("workflowMessages", messageData) # Clear cache to ensure fresh data self._clearTableCache("workflowMessages") # Convert to ChatMessage model return ChatMessage( id=createdMessage["id"], workflowId=createdMessage["workflowId"], parentMessageId=createdMessage.get("parentMessageId"), agentName=createdMessage.get("agentName"), documents=[ChatDocument(**doc) for doc in createdMessage.get("documents", [])], documentsLabel=createdMessage.get("documentsLabel"), # <-- FIX: ensure label is set message=createdMessage.get("message"), role=createdMessage.get("role", "assistant"), status=createdMessage.get("status", "step"), sequenceNr=len(workflow.messages) + 1, # Use messages list length for sequence number publishedAt=createdMessage.get("publishedAt", get_utc_timestamp()), stats=ChatStat(**createdMessage.get("stats", {})) if createdMessage.get("stats") else None, # CRITICAL FIX: Include the progress fields in the ChatMessage object 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") ) except Exception as e: logger.error(f"Error creating workflow message: {str(e)}") return None def updateWorkflowMessage(self, messageId: str, messageData: Dict[str, Any]) -> Dict[str, Any]: """Updates a workflow message if user has access to the workflow.""" try: logger.debug(f"Updating message {messageId} in database") # Ensure messageId is provided if not messageId: logger.error("No messageId provided for updateWorkflowMessage") raise ValueError("messageId cannot be empty") # Check if message exists in database messages = self.db.getRecordset("workflowMessages", recordFilter={"id": messageId}) if not messages: logger.warning(f"Message with ID {messageId} does not exist in database") # If message doesn't exist but we have workflowId, create it if "workflowId" in messageData: workflowId = messageData.get("workflowId") # Check workflow access workflow = self.getWorkflow(workflowId) if not workflow: raise PermissionError(f"No access to workflow {workflowId}") if not self._canModify("workflows", workflowId): raise PermissionError(f"No permission to modify workflow {workflowId}") logger.info(f"Creating new message with ID {messageId} for workflow {workflowId}") return self.db.recordCreate("workflowMessages", messageData) else: logger.error(f"Workflow ID missing for new message {messageId}") return None # Update existing message existingMessage = messages[0] # Check workflow access workflowId = existingMessage.get("workflowId") workflow = self.getWorkflow(workflowId) if not workflow: raise PermissionError(f"No access to workflow {workflowId}") if not self._canModify("workflows", workflowId): raise PermissionError(f"No permission to modify workflow {workflowId}") # Ensure required fields present for key in ["role", "agentName"]: if key not in messageData and key not in existingMessage: messageData[key] = "assistant" if key == "role" else "" # Ensure ID is in the dataset if 'id' not in messageData: messageData['id'] = messageId # Convert createdAt to startedAt if needed if "createdAt" in messageData and "startedAt" not in messageData: messageData["startedAt"] = messageData["createdAt"] del messageData["createdAt"] # Update the message updatedMessage = self.db.recordModify("workflowMessages", messageId, messageData) if updatedMessage: logger.debug(f"Message {messageId} updated successfully") # Clear cache to ensure fresh data self._clearTableCache("workflowMessages") else: logger.warning(f"Failed to update message {messageId}") return 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 deleteWorkflowMessage(self, workflowId: str, messageId: str) -> bool: """Deletes a workflow message if user has access to the workflow.""" try: # Check workflow access workflow = self.getWorkflow(workflowId) if not workflow: logger.warning(f"No access to workflow {workflowId}") return False if not self._canModify("workflows", workflowId): raise PermissionError(f"No permission to modify workflow {workflowId}") # Check if the message exists messages = self.getWorkflowMessages(workflowId) message = next((m for m in messages if m.get("id") == messageId), None) if not message: logger.warning(f"Message {messageId} for workflow {workflowId} not found") return False # Delete the message from the database success = self.db.recordDelete("workflowMessages", messageId) # Clear cache to ensure fresh data self._clearTableCache("workflowMessages") return success except Exception as e: logger.error(f"Error deleting message {messageId}: {str(e)}") return False def deleteFileFromMessage(self, workflowId: str, messageId: str, fileId: str) -> bool: """Removes a file reference from a message if user has access.""" try: # Check workflow access workflow = self.getWorkflow(workflowId) if not workflow: logger.warning(f"No access to workflow {workflowId}") return False if not self._canModify("workflows", workflowId): raise PermissionError(f"No permission to modify workflow {workflowId}") logger.debug(f"Removing file {fileId} from message {messageId} in workflow {workflowId}") # Get all workflow messages allMessages = self.getWorkflowMessages(workflowId) logger.debug(f"Workflow {workflowId} has {len(allMessages)} messages") # Try different approaches to find the message message = None # Exact match message = next((m for m in allMessages if m.get("id") == messageId), None) # Case-insensitive match if not message and isinstance(messageId, str): message = next((m for m in allMessages if isinstance(m.get("id"), str) and m.get("id").lower() == messageId.lower()), None) # Partial match (starts with) if not message and isinstance(messageId, str): message = next((m for m in allMessages if isinstance(m.get("id"), str) and m.get("id").startswith(messageId)), None) if not message: logger.warning(f"Message {messageId} not found in workflow {workflowId}") return False # Log the found message logger.debug(f"Found message: {message.get('id')}") # Check if message has documents if "documents" not in message or not message["documents"]: logger.warning(f"No documents in message {messageId}") return False # Log existing documents documents = message.get("documents", []) logger.debug(f"Message has {len(documents)} documents") # Create a new list of documents without the one to delete updatedDocuments = [] 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: removed = True logger.debug(f"Found file to remove: docId={docId}, fileId={fileIdValue}") else: updatedDocuments.append(doc) if not removed: logger.warning(f"No matching file {fileId} found in message {messageId}") return False # Update message with modified documents array messageUpdate = { "documents": updatedDocuments } # Apply the update directly to the database updated = self.db.recordModify("workflowMessages", message["id"], messageUpdate) if updated: logger.debug(f"Successfully removed file {fileId} from message {messageId}") return True else: logger.warning(f"Failed to update message {messageId} in database") return False except Exception as e: logger.error(f"Error removing file {fileId} from message {messageId}: {str(e)}") return False # Workflow Logs def getWorkflowLogs(self, workflowId: str) -> List[ChatLog]: """Returns logs for a workflow if user has access to the workflow.""" # Check workflow access first workflow = self.getWorkflow(workflowId) if not workflow: return [] # Get logs for this workflow logs = self.db.getRecordset("workflowLogs", recordFilter={"workflowId": workflowId}) # Sort logs by timestamp (Unix timestamps) logs.sort(key=lambda x: float(x.get("timestamp", 0))) return [ChatLog(**log) for log in logs] def updateWorkflowStats(self, workflowId: str, bytesSent: int = 0, bytesReceived: int = 0) -> bool: """Updates workflow statistics during execution with incremental values.""" try: # Get current workflow workflow = self.getWorkflow(workflowId) if not workflow: logger.error(f"Workflow {workflowId} not found for stats update") return False if not self._canModify("workflows", workflowId): logger.error(f"No permission to update workflow {workflowId} stats") return False # Get current stats - ensure we have proper defaults if workflow.stats: currentStats = workflow.stats.dict() # Ensure all required fields exist currentStats.setdefault("bytesSent", 0) currentStats.setdefault("bytesReceived", 0) currentStats.setdefault("tokenCount", 0) currentStats.setdefault("processingTime", 0) else: currentStats = { "bytesSent": 0, "bytesReceived": 0, "tokenCount": 0, "processingTime": 0 } # Calculate processing time as duration since workflow start using Unix timestamps workflow = self.getWorkflow(workflowId) if workflow and workflow.startedAt: try: # Parse start time as Unix timestamp (handle both old ISO format and new Unix format) start_time_str = workflow.startedAt try: # Try to parse as Unix timestamp first start_time = int(float(start_time_str)) except ValueError: # If that fails, try to parse as ISO format and convert to Unix try: # Handle ISO format timestamps (for backward compatibility) if start_time_str.endswith('Z'): start_time_str = start_time_str.replace('Z', '+00:00') dt = datetime.fromisoformat(start_time_str) start_time = int(dt.timestamp()) except: # If all parsing fails, use current time logger.warning(f"Could not parse start time: {start_time_str}, using current time") start_time = int(get_utc_timestamp()) current_time = int(get_utc_timestamp()) processing_time = current_time - start_time # Ensure processing time is reasonable (not negative or extremely large) if processing_time < 0: logger.warning(f"Negative processing time calculated: {processing_time}, using 0") processing_time = 0 elif processing_time > 86400 * 365: # More than 1 year logger.warning(f"Unreasonably large processing time: {processing_time}, using 0") processing_time = 0 except Exception as e: logger.warning(f"Error calculating processing time: {str(e)}") processing_time = currentStats.get("processingTime", 0) or 0 else: # Fallback to existing processing time or 0 processing_time = currentStats.get("processingTime", 0) or 0 # Update stats with incremental values - ensure no None values current_bytes_sent = currentStats.get("bytesSent", 0) or 0 current_bytes_received = currentStats.get("bytesReceived", 0) or 0 currentStats["bytesSent"] = current_bytes_sent + bytesSent currentStats["bytesReceived"] = current_bytes_received + bytesReceived currentStats["tokenCount"] = currentStats["bytesSent"] + currentStats["bytesReceived"] currentStats["processingTime"] = processing_time # Update workflow in database self.db.recordModify("workflows", workflowId, { "stats": currentStats }) # Log to stats table stats_record = { "timestamp": get_utc_timestamp(), "workflowId": workflowId, "bytesSent": bytesSent, "bytesReceived": bytesReceived, "tokenCount": bytesSent + bytesReceived, "processingTime": processing_time } # Create stats record in database self.db.recordCreate("stats", stats_record) # logger.debug(f"Updated workflow {workflowId} stats: {currentStats}") # logger.debug(f"Logged stats record: {stats_record}") return True except Exception as e: logger.error(f"Error updating workflow stats: {str(e)}") return False def createWorkflowLog(self, logData: Dict[str, Any]) -> ChatLog: """Creates a log entry for a workflow if user has access.""" # Check workflow access workflowId = logData.get("workflowId") if not workflowId: logger.error("No workflowId provided for createWorkflowLog") return None workflow = self.getWorkflow(workflowId) if not workflow: logger.warning(f"No access to workflow {workflowId}") return None if not self._canModify("workflows", workflowId): logger.warning(f"No permission to modify workflow {workflowId}") return None # Make sure required fields are present if "timestamp" not in logData: logData["timestamp"] = get_utc_timestamp() # Add status information if not present if "status" not in logData and "type" in logData: if logData["type"] == "error": logData["status"] = "error" else: logData["status"] = "running" # Add progress information if not present if "progress" not in logData: # Default progress values based on log type if logData.get("type") == "info": logData["progress"] = 50 # Default middle progress elif logData.get("type") == "error": logData["progress"] = -1 # Error state elif logData.get("type") == "warning": logData["progress"] = 50 # Default middle progress # Validate log data against ChatLog model try: log_model = ChatLog(**logData) except Exception as e: logger.error(f"Invalid log data: {str(e)}") return None # Create log in database createdLog = self.db.recordCreate("workflowLogs", log_model.to_dict()) # Clear cache to ensure fresh data self._clearTableCache("workflowLogs") # Return validated ChatLog instance return ChatLog(**createdLog) def loadWorkflowState(self, workflowId: str) -> Optional[ChatWorkflow]: """Loads workflow state if user has access.""" try: # Check workflow access workflow = self.getWorkflow(workflowId) if not workflow: return None logger.debug(f"Loaded base workflow {workflowId} from database") # Load messages messages = self.getWorkflowMessages(workflowId) # Messages are already sorted by publishedAt in getWorkflowMessages messageCount = len(messages) logger.debug(f"Loaded {messageCount} messages for workflow {workflowId}") # Log document counts for each message for msg in messages: docCount = len(msg.documents) if hasattr(msg, 'documents') else 0 if docCount > 0: logger.debug(f"Message {msg.id} has {docCount} documents loaded from database") # Load logs logs = self.getWorkflowLogs(workflowId) # Logs are already sorted by timestamp in getWorkflowLogs # Create a new ChatWorkflow object with loaded messages and logs return ChatWorkflow( id=workflow.id, status=workflow.status, name=workflow.name, currentRound=workflow.currentRound, lastActivity=workflow.lastActivity, startedAt=workflow.startedAt, logs=logs, messages=messages, stats=workflow.stats, mandateId=workflow.mandateId ) except Exception as e: logger.error(f"Error loading workflow state: {str(e)}") return None # Workflow Actions async def workflowStart(self, currentUser: User, userInput: UserInputRequest, workflowId: Optional[str] = None) -> ChatWorkflow: """ Starts a new workflow or continues an existing one. Args: userInput: The user input request containing workflow initialization data workflowId: Optional ID of an existing workflow to continue Returns: ChatWorkflow object representing the started/continued workflow """ try: # Get current timestamp currentTime = get_utc_timestamp() if workflowId: # Continue existing workflow - load complete state including messages workflow = self.loadWorkflowState(workflowId) if not workflow: raise ValueError(f"Workflow {workflowId} not found") # Check if workflow is currently running and stop it first if workflow.status == "running": logger.info(f"Stopping running workflow {workflowId} before processing new prompt") # Stop the running workflow workflow.status = "stopped" workflow.lastActivity = currentTime self.updateWorkflow(workflowId, { "status": "stopped", "lastActivity": currentTime }) # Add log entry for workflow stop self.createWorkflowLog({ "workflowId": workflowId, "message": "Workflow stopped for new prompt", "type": "info", "status": "stopped", "progress": 100 }) # Wait a moment for any running processes to detect the stop await asyncio.sleep(0.1) # Update workflow - increment round for existing workflows newRound = workflow.currentRound + 1 self.updateWorkflow(workflowId, { "status": "running", # Set status back to running for resumed workflows "lastActivity": currentTime, "currentRound": newRound }) # Reload workflow object to get updated currentRound from database workflow = self.loadWorkflowState(workflowId) if not workflow: raise ValueError(f"Failed to reload workflow {workflowId} after update") # Add log entry for workflow resumption self.createWorkflowLog({ "workflowId": workflowId, "message": f"Workflow resumed (round {workflow.currentRound})", "type": "info", "status": "running", "progress": 0 }) else: # Create new workflow workflowData = { "name": "New Workflow", # Default name since UserInputRequest doesn't have a name field "status": "running", "startedAt": currentTime, "lastActivity": currentTime, "currentRound": 0, # Default value, will be set to 1 in workflowStart() "currentTask": 0, "currentAction": 0, "totalTasks": 0, "totalActions": 0, "mandateId": self.mandateId, "messageIds": [], "stats": { "processingTime": None, "tokenCount": None, "bytesSent": None, "bytesReceived": None, "successRate": None, "errorCount": None } } # Create workflow workflow = self.createWorkflow(workflowData) # Set currentRound to 1 for new workflows workflow.currentRound = 1 self.updateWorkflow(workflow.id, {"currentRound": 1}) # Initialize stats for the new workflow self.updateWorkflowStats(workflow.id, bytesSent=0, bytesReceived=0) # Remove the 'Workflow started' log entry # Start workflow processing from modules.workflow.managerWorkflow import WorkflowManager workflowManager = WorkflowManager(self, currentUser) # Start the workflow processing asynchronously # The workflow will be updated with progress data during execution asyncio.create_task(workflowManager.workflowProcess(userInput, workflow)) return workflow except Exception as e: logger.error(f"Error starting workflow: {str(e)}") raise async def workflowStop(self, workflowId: str) -> ChatWorkflow: """ Stops a running workflow (State 8: Workflow Stopped). Args: workflowId: ID of the workflow to stop Returns: Updated ChatWorkflow object """ try: # Load workflow state workflow = self.getWorkflow(workflowId) if not workflow: raise ValueError(f"Workflow {workflowId} not found") # Update workflow status workflow.status = "stopped" workflow.lastActivity = get_utc_timestamp() # Update in database self.updateWorkflow(workflowId, { "status": "stopped", "lastActivity": workflow.lastActivity }) # Add log entry self.createWorkflowLog({ "workflowId": workflowId, "message": "Workflow stopped", "type": "warning", "status": "stopped", "progress": 100 }) return workflow except Exception as e: logger.error(f"Error stopping workflow: {str(e)}") raise def getInterface(currentUser: Optional[User] = None) -> 'ChatObjects': """ Returns a ChatObjects instance for the current user. Handles initialization of database and records. """ if not currentUser: raise ValueError("Invalid user context: user is required") # Create context key contextKey = f"{currentUser.mandateId}_{currentUser.id}" # Create new instance if not exists if contextKey not in _chatInterfaces: _chatInterfaces[contextKey] = ChatObjects(currentUser) return _chatInterfaces[contextKey]