content: import asyncio
  import json
  from datetime import datetime, timezone
  from typing import AsyncGenerator, List, Optional, Dict, Union
  from sqlalchemy.exc import IntegrityError, SQLAlchemyError
  from sqlalchemy.orm import Session
  from uuid6 import uuid7
  
  from app.modules.code_provider.code_provider_service import CodeProviderService
  from app.modules.utils.logger import setup_logger
  from app.modules.conversations.conversation.conversation_model import (
      Conversation,
      ConversationStatus,
      Visibility,
  )
  from app.modules.conversations.conversation.conversation_schema import (
      ChatMessageResponse,
      ConversationAccessType,
      ConversationInfoResponse,
      CreateConversationRequest,
  )
  from app.modules.conversations.message.message_model import (
      MessageType,
  )
  from app.modules.conversations.message.message_schema import (
      MessageRequest,
      MessageResponse,
      NodeContext,
  )
  from app.modules.intelligence.agents.custom_agents.custom_agents_service import (
      CustomAgentService,
  )
  from app.modules.intelligence.agents.agents_service import AgentsService
  from app.modules.intelligence.agents.chat_agent import ChatContext
  from app.modules.intelligence.memory.chat_history_service import ChatHistoryService
  from app.modules.intelligence.provider.provider_service import ProviderService
  from app.modules.projects.projects_service import ProjectService
  from app.modules.users.user_service import UserService
  from app.modules.utils.posthog_helper import PostHogClient
  from app.modules.intelligence.prompts.prompt_service import PromptService
  from app.modules.intelligence.tools.tool_service import ToolService
  from app.modules.media.media_service import MediaService
  from app.modules.conversations.session.session_service import SessionService
  from app.modules.conversations.utils.redis_streaming import RedisStreamManager
  from app.celery.celery_app import celery_app
  from .conversation_store import ConversationStore, StoreError
  from ..message.message_store import MessageStore
  
  logger = setup_logger(__name__)
  
  
  class ConversationServiceError(Exception):
      pass
  
  
  class ConversationNotFoundError(ConversationServiceError):
      pass
  
  
  class MessageNotFoundError(ConversationServiceError):
      pass
  
  
  class AccessTypeNotFoundError(ConversationServiceError):
      pass
  
  
  class AccessTypeReadError(ConversationServiceError):
      pass
  
  
  class ConversationService:
      def __init__(
          self,
          db: Session,
          user_id: str,
          user_email: str,
          conversation_store: ConversationStore,
          message_store: MessageStore,
          project_service: ProjectService,
          history_manager: ChatHistoryService,
          provider_service: ProviderService,
          tools_service: ToolService,
          promt_service: PromptService,
          agent_service: AgentsService,
          custom_agent_service: CustomAgentService,
          media_service: MediaService,
          session_service: SessionService = None,
          redis_manager: RedisStreamManager = None,
      ):
          self.db = db
          self.user_id = user_id
          self.user_email = user_email
          self.conversation_store = conversation_store
          self.message_store = message_store
          self.project_service = project_service
          self.history_manager = history_manager
          self.provider_service = provider_service
          self.tool_service = tools_service
          self.prompt_service = promt_service
          self.agent_service = agent_service
          self.custom_agent_service = custom_agent_service
          self.media_service = media_service
          # Dependency injection for stop_generation
          self.session_service = session_service or SessionService()
          self.redis_manager = redis_manager or RedisStreamManager()
          self.celery_app = celery_app
  
      @classmethod
      def create(
          cls,
          conversation_store: ConversationStore,
          message_store: MessageStore,
          db: Session,
          user_id: str,
          user_email: str,
      ):
          project_service = ProjectService(db)
          history_manager = ChatHistoryService(db)
          provider_service = ProviderService(db, user_id)
          tool_service = ToolService(db, user_id)
          prompt_service = PromptService(db)
          agent_service = AgentsService(
              db, provider_service, prompt_service, tool_service
          )
          custom_agent_service = CustomAgentService(db, provider_service, tool_service)
          media_service = MediaService(db)
          session_service = SessionService()
          redis_manager = RedisStreamManager()
  
          return cls(
              db,
              user_id,
              user_email,
              conversation_store,
              message_store,
              project_service,
              history_manager,
              provider_service,
              tool_service,
              prompt_service,
              agent_service,
              custom_agent_service,
              media_service,
              session_service,
              redis_manager,
          )
  
      async def check_conversation_access(
          self, conversation_id: str, user_email: str, firebase_user_id: str = None
      ) -> str:
          if not user_email:
              return ConversationAccessType.WRITE
  
          # Use Firebase user ID directly if available, otherwise fall back to email lookup
          if firebase_user_id:
              user_id = firebase_user_id
          else:
              user_service = UserService(self.sql_db)
              user_id = user_service.get_user_id_by_email(user_email)
  
          # Retrieve the conversation
          conversation = await self.conversation_store.get_by_id(conversation_id)
  
          if not conversation:
              logger.warning(f"Conversation {conversation_id} not found in database")
              return (
                  ConversationAccessType.NOT_FOUND
              )  # Return 'not found' if conversation doesn't exist
  
          if not conversation.visibility:
              conversation.visibility = Visibility.PRIVATE
  
          if user_id == conversation.user_id:  # Check if the user is the creator
              return ConversationAccessType.WRITE  # Creator always has write access
  
          if conversation.visibility == Visibility.PUBLIC:
              return ConversationAccessType.READ  # Public users get read access
  
          # Check if the conversation is shared
          if conversation.shared_with_emails:
              user_service = UserService(self.sql_db)
              shared_user_ids = user_service.get_user_ids_by_emails(
                  conversation.shared_with_emails
              )
              if shared_user_ids is None:
                  logger.warning(
                      "Failed to get user IDs for shared emails, returning NOT_FOUND"
                  )
                  return ConversationAccessType.NOT_FOUND
              # Check if the current user ID is in the shared user IDs
              if user_id in shared_user_ids:
                  return ConversationAccessType.READ  # Shared users can only read
              else:
                  return ConversationAccessType.NOT_FOUND
  
          return ConversationAccessType.NOT_FOUND
  
      async def create_conversation(
          self,
          conversation: CreateConversationRequest,
          user_id: str,
          hidden: bool = False,
      ) -> tuple[str, str]:
          try:
              if not await self.agent_service.validate_agent_id(
                  user_id, conversation.agent_ids[0]
              ):
                  raise ConversationServiceError(
                      f"Invalid agent_id: {conversation.agent_ids[0]}"
                  )
  
              project_name = await self.project_service.get_project_name(
                  conversation.project_ids
              )
  
              title = (
                  conversation.title.strip().replace("Untitled", project_name)
                  if conversation.title
                  else project_name
              )
  
              conversation_id = await self._create_conversation_record(
                  conversation, title, user_id, hidden
              )
  
              asyncio.create_task(
                  CodeProviderService(self.db).get_project_structure_async(
                      conversation.project_ids[0]
                  )
              )
  
              await self._add_system_message(conversation_id, project_name, user_id)
  
              return conversation_id, "Conversation created successfully."
          except IntegrityError as e:
              logger.exception("IntegrityError in create_conversation", user_id=user_id)
              raise ConversationServiceError(
                  "Failed to create conversation due to a database integrity error."
              ) from e
          except Exception as e:
              logger.exception("Unexpected error in create_conversation", user_id=user_id)
              raise ConversationServiceError(
                  "An unexpected error occurred while creating the conversation."
              ) from e
  
      async def _create_conversation_record(
          self,
          conversation: CreateConversationRequest,
          title: str,
          user_id: str,
          hidden: bool = False,
      ) -> str:
          conversation_id = str(uuid7())
          new_conversation = Conversation(
              id=conversation_id,
              user_id=user_id,
              title=title,
              status=ConversationStatus.ARCHIVED if hidden else ConversationStatus.ACTIVE,
              project_ids=conversation.project_ids,
              agent_ids=conversation.agent_ids,
              created_at=datetime.now(timezone.utc),
              updated_at=datetime.now(timezone.utc),
          )
          await self.conversation_store.create(new_conversation)
  
          logger.info(
              f"Project id : {conversation.project_ids[0]} Created new conversation with ID: {conversation_id}, title: {title}, user_id: {user_id}, agent_id: {conversation.agent_ids[0]}, hidden: {hidden}"
          )
          return conversation_id
  
      async def _add_system_message(
          self, conversation_id: str, project_name: str, user_id: str
      ):
          content = f"You can now ask questions about the {project_name} repository."
          try:
              self.history_manager.add_message_chunk(
                  conversation_id, content, MessageType.SYSTEM_GENERATED, user_id
              )
              self.history_manager.flush_message_buffer(
                  conversation_id, MessageType.SYSTEM_GENERATED, user_id
              )
              logger.info(
                  f"Added system message to conversation {conversation_id} for user {user_id}"
              )
          except Exception as e:
              logger.exception(
                  f"Failed to add system message to conversation {conversation_id}",
                  conversation_id=conversation_id,
                  user_id=user_id,
              )
              raise ConversationServiceError(
                  "Failed to add system message to the conversation."
              ) from e
  
      async def store_message(
          self,
          conversation_id: str,
          message: MessageRequest,
          message_type: MessageType,
          user_id: str,
          stream: bool = True,
      ) -> AsyncGenerator[ChatMessageResponse, None]:
          try:
              logger.info(
                  f"DEBUG: store_message called with message.attachment_ids: {message.attachment_ids}"
              )
              access_level = await self.check_conversation_access(
                  conversation_id, self.user_email, user_id
              )
              if access_level == ConversationAccessType.READ:
                  raise AccessTypeReadError("Access denied.")
              self.history_manager.add_message_chunk(
                  conversation_id, message.content, message_type, user_id
              )
              message_id = self.history_manager.flush_message_buffer(
                  conversation_id, message_type, user_id
              )
              logger.info(f"Stored message in conversation {conversation_id}")
  
              # Handle attachments if present
              if message_type == MessageType.HUMAN and message.attachment_ids:
                  try:
                      await self.media_service.update_message_attachments(
                          message_id, message.attachment_ids
                      )
                      logger.info(
                          f"Linked {len(message.attachment_ids)} attachments to message {message_id}"
                      )
                  except Exception:
                      logger.exception(
                          f"Failed to link attachments to message {message_id}",
                          message_id=message_id,
                          conversation_id=conversation_id,
                      )
                      # Continue processing even if attachment linking fails
  
              if message_type == MessageType.HUMAN:
                  conversation = await self._get_conversation_with_message_count(
                      conversation_id
                  )
                  if not conversation:
                      raise ConversationNotFoundError(
                          f"Conversation with id {conversation_id} not found"
                      )
  
                  # Check if this is the first human message
                  if conversation.human_message_count == 1:
                      new_title = await self._generate_title(
                          conversation, message.content
                      )
                      await self._update_conversation_title(conversation_id, new_title)
  
                  project_id = (
                      conversation.project_ids[0] if conversation.project_ids else None
                  )
                  if not project_id:
                      raise ConversationServiceError(
                          "No project associated with this conversation"
                      )
  
                  if stream:
                      async for chunk in self._generate_and_stream_ai_response(
                          message.content,
                          conversation_id,
                          user_id,
                          message.node_ids,
                          message.attachment_ids,
                      ):
                          yield chunk
                  else:
                      full_message = ""
                      all_citations = []
                      async for chunk in self._generate_and_stream_ai_response(
                          message.content,
                          conversation_id,
                          user_id,
                          message.node_ids,
                          message.attachment_ids,
                      ):
                          full_message += chunk.message
                          all_citations = all_citations + chunk.citations
  
                      yield ChatMessageResponse(
                          message=full_message, citations=all_citations, tool_calls=[]
                      )
  
          except AccessTypeReadError:
              raise
          except Exception as e:
              logger.exception(
                  f"Error in store_message for conversation {conversation_id}",
                  conversation_id=conversation_id,
                  user_id=user_id,
              )
              raise ConversationServiceError(
                  "Failed to store message or generate AI response."
              ) from e
  
      async def _get_conversation_with_message_count(
          self, conversation_id: str
      ) -> Conversation:
          return await self.conversation_store.get_with_message_count(conversation_id)
  
      async def _generate_title(
          self, conversation: Conversation, message_content: str
      ) -> str:
          agent_type = conversation.agent_ids[0]
  
          prompt = (
              "Given an agent type '{agent_type}' and an initial message '{message}', "
              "generate a concise and relevant title for a conversation. "
              "The title should be no longer than 50 characters. Only return title string, do not wrap in quotes."
          ).format(agent_type=agent_type, message=message_content)
  
          messages = [
              {
                  "role": "system",
                  "content": "You are a conversation title generator that creates concise and relevant titles.",
              },
              {"role": "user", "content": prompt},
          ]
          generated_title: str = await self.provider_service.call_llm(
              messages=messages, config_type="chat"
          )  # type: ignore
  
          if len(generated_title) > 50:
              generated_title = generated_title[:50].strip() + "..."
          return generated_title
  
      async def _update_conversation_title(self, conversation_id: str, new_title: str):
          await self.conversation_store.update_title(conversation_id, new_title)
  
      async def regenerate_last_message(
          self,
          conversation_id: str,
          user_id: str,
          node_ids: List[NodeContext] = [],
          stream: bool = True,
      ) -> AsyncGenerator[ChatMessageResponse, None]:
          try:
              access_level = await self.check_conversation_access(
                  conversation_id, self.user_email, user_id
              )
              if access_level != ConversationAccessType.WRITE:
                  raise AccessTypeReadError(
                      "Access denied. Only conversation creators can regenerate messages."
                  )
              last_human_message = await self._get_last_human_message(conversation_id)
              if not last_human_message:
                  raise MessageNotFoundError("No human message found to regenerate from")
  
              # Get attachment IDs from the last human message
              attachment_ids = None
              if last_human_message.has_attachments:
                  try:
                      attachments = await self.media_service.get_message_attachments(
                          last_human_message.id, include_download_urls=False
                      )
                      # Extract only image attachment IDs for multimodal processing
                      from app.modules.media.media_model import AttachmentType
  
                      attachment_ids = [
                          att.id
                          for att in attachments
                          if att.attachment_type == AttachmentType.IMAGE
                      ]
                      if attachment_ids:
                          logger.info(
                              f"Found {len(attachment_ids)} image attachments for regeneration: {attachment_ids}"
                          )
                      else:
                          logger.info("No image attachments found in last human message")
                  except Exception as e:
                      logger.warning(
                          f"Failed to retrieve attachments for message {last_human_message.id}: {e}"
                      )
                      attachment_ids = None
  
              await self._archive_subsequent_messages(
                  conversation_id, last_human_message.created_at
              )
              PostHogClient().send_event(
                  user_id,
                  "regenerate_conversation_event",
                  {"conversation_id": conversation_id},
              )
  
              if stream:
                  async for chunk in self._generate_and_stream_ai_response(
                      last_human_message.content,
                      conversation_id,
                      user_id,
                      node_ids,
                      attachment_ids,
                  ):
                      yield chunk
              else:
                  full_message = ""
                  all_citations = []
  
                  async for chunk in self._generate_and_stream_ai_response(
                      last_human_message.content,
                      conversation_id,
                      user_id,
                      node_ids,
                      attachment_ids,
                  ):
                      full_message += chunk.message
                      all_citations = all_citations + chunk.citations
  
                  yield ChatMessageResponse(
                      message=full_message, citations=all_citations, tool_calls=[]
                  )
  
          except AccessTypeReadError:
              raise
          except MessageNotFoundError as e:
              logger.warning(
                  f"No message to regenerate in conversation {conversation_id}: {e}"
              )
              raise
          except Exception as e:
              logger.exception(
                  f"Error in regenerate_last_message for conversation {conversation_id}",
                  conversation_id=conversation_id,
                  user_id=user_id,
              )
              raise ConversationServiceError("Failed to regenerate last message.") from e
  
      async def regenerate_last_message_background(
          self,
          conversation_id: str,
          node_ids: Optional[List[str]] = None,
          attachment_ids: List[str] = [],
      ) -> AsyncGenerator[ChatMessageResponse, None]:
          """Background version of regenerate_last_message for Celery task execution"""
          try:
              # Access control validation
              access_level = await self.check_conversation_access(
                  conversation_id, self.user_email, self.user_id
              )
              if access_level != ConversationAccessType.WRITE:
                  raise AccessTypeReadError(
                      "Access denied. Only conversation creators can regenerate messages."
                  )
  
              # Get last human message (already validated by background task caller)
              last_human_message = await self._get_last_human_message(conversation_id)
              if not last_human_message:
                  raise MessageNotFoundError("No human message found to regenerate from")
  
              # Archive subsequent messages
              await self._archive_subsequent_messages(
                  conversation_id, last_human_message.created_at
              )
  
              # PostHog analytics
              PostHogClient().send_event(
                  self.user_id,
                  "regenerate_conversation_event",
                  {"conversation_id": conversation_id},
              )
  
              # Convert string node_ids to NodeContext objects for compatibility
              node_contexts = []
              if node_ids:
                  node_contexts = [NodeContext(node_id=node_id) for node_id in node_ids]
  
              # Execute AI response generation with existing logic
              async for chunk in self._generate_and_stream_ai_response(
                  last_human_message.content,
                  conversation_id,
                  self.user_id,
                  node_contexts,
                  attachment_ids,
              ):
                  yield chunk
  
          except (AccessTypeReadError, MessageNotFoundError):
              logger.exception(
                  f"Background regeneration error for {conversation_id}",
                  conversation_id=conversation_id,
                  user_id=self.user_id,
              )
              raise
          except Exception as e:
              logger.exception(
                  f"Background regeneration failed for {conversation_id}",
                  conversation_id=conversation_id,
                  user_id=self.user_id,
              )
              raise ConversationServiceError(f"Failed to regenerate message: {str(e)}")
  
      async def _get_last_human_message(self, conversation_id: str):
          message = await self.message_store.get_last_human_message(conversation_id)
  
          if not message:
              logger.warning(f"No human message found in conversation {conversation_id}")
          return message
  
      async def _archive_subsequent_messages(
          self, conversation_id: str, timestamp: datetime
      ):
          try:
              await self.message_store.archive_messages_after(conversation_id, timestamp)
  
              logger.info(
                  f"Archived subsequent messages in conversation {conversation_id}"
              )
          except Exception as e:
              logger.exception(
                  f"Failed to archive messages in conversation {conversation_id}",
                  conversation_id=conversation_id,
                  user_id=self.user_id,
              )
              raise ConversationServiceError(
                  "Failed to archive subsequent messages."
              ) from e
  
      def parse_str_to_message(self, chunk: str) -> ChatMessageResponse:
          try:
              data = json.loads(chunk)
          except json.JSONDecodeError as e:
              logger.exception("Failed to parse chunk as JSON")
              raise ConversationServiceError("Failed to parse AI response") from e
  
          # Extract the 'message' and 'citations'
          message: str = data.get("message", "")
          citations: List[str] = data.get("citations", [])
          tool_calls: List[dict] = data.get("tool_calls", [])
  
          return ChatMessageResponse(
              message=message, citations=citations, tool_calls=tool_calls
          )
  
      async def _generate_and_stream_ai_response(
          self,
          query: str,
          conversation_id: str,
          user_id: str,
          node_ids: List[NodeContext],
          attachment_ids: Optional[List[str]] = None,
      ) -> AsyncGenerator[ChatMessageResponse, None]:
          conversation = await self.conversation_store.get_by_id(conversation_id)
          if not conversation:
              raise ConversationNotFoundError(
                  f"Conversation with id {conversation_id} not found"
              )
  
          agent_id = conversation.agent_ids[0]
          project_id = conversation.project_ids[0] if conversation.project_ids else None
  
          try:
              history = self.history_manager.get_session_history(user_id, conversation_id)
              validated_history = [
                  (f"{msg.type}: {msg.content}" if msg.content else msg)
                  for msg in history
              ]
  
          except Exception:
              raise ConversationServiceError("Failed to get chat history")
  
          try:
              type = await self.agent_service.validate_agent_id(user_id, str(agent_id))
              if type is None:
                  raise ConversationServiceError(f"Invalid agent_id {agent_id}")
  
              project_name = await self.project_service.get_project_name(
                  project_ids=[project_id]
              )
  
              # Prepare multimodal context - use current message attachments if available
              image_attachments = None
              if attachment_ids:
                  image_attachments = await self._prepare_attachments_as_images(
                      attachment_ids
                  )
  
              # Also get context images from recent conversation history
              context_images = await self._prepare_conversation_context_images(
                  conversation_id
              )
  
              logger.info(
                  f"conversation_id: {conversation_id} Running agent {agent_id} with query: {query}"
              )
  
              if image_attachments or context_images:
                  logger.info(
                      f"Multimodal context: {len(image_attachments) if image_attachments else 0} current images, {len(context_images) if context_images else 0} context images"
                  )
  
              if type == "CUSTOM_AGENT":
                  res = (
                      await self.agent_service.custom_agent_service.execute_agent_runtime(
                          user_id,
                          ChatContext(
                              project_id=str(project_id),
                              project_name=project_name,
                              curr_agent_id=str(agent_id),
                              history=validated_history[-12:],
                              node_ids=[node.node_id for node in node_ids],
                              query=query,
                          ),
                      )
                  )
                  async for chunk in res:
                      self.history_manager.add_message_chunk(
                          conversation_id,
                          chunk.response,
                          MessageType.AI_GENERATED,
                          citations=chunk.citations,
                      )
                      yield ChatMessageResponse(
                          message=chunk.response,
                          citations=chunk.citations,
                          tool_calls=[
                              tool_call.model_dump_json()
                              for tool_call in chunk.tool_calls
                          ],
                      )
                  self.history_manager.flush_message_buffer(
                      conversation_id, MessageType.AI_GENERATED
                  )
              else:
                  # Create enhanced ChatContext with multimodal support
                  nodes = [] if node_ids is None else [node.node_id for node in node_ids]
                  chat_context = ChatContext(
                      project_id=str(project_id),
                      project_name=project_name,
                      curr_agent_id=str(agent_id),
                      history=validated_history[-8:],
                      node_ids=nodes,
                      query=query,
                      image_attachments=image_attachments,
                      context_images=context_images,
                  )
  
                  res = self.agent_service.execute_stream(chat_context)
  
                  async for chunk in res:
                      self.history_manager.add_message_chunk(
                          conversation_id,
                          chunk.response,
                          MessageType.AI_GENERATED,
                          citations=chunk.citations,
                      )
                      yield ChatMessageResponse(
                          message=chunk.response,
                          citations=chunk.citations,
                          tool_calls=[
                              tool_call.model_dump_json()
                              for tool_call in chunk.tool_calls
                          ],
                      )
                  self.history_manager.flush_message_buffer(
                      conversation_id, MessageType.AI_GENERATED
                  )
  
              logger.info(
                  f"Generated and streamed AI response for conversation {conversation.id} for user {user_id} using agent {agent_id}"
              )
          except Exception as e:
              logger.exception(
                  f"Failed to generate and stream AI response for conversation {conversation.id}",
                  conversation_id=conversation.id,
                  user_id=user_id,
              )
              raise ConversationServiceError(
                  "Failed to generate and stream AI response."
              ) from e
  
      async def _generate_and_stream_ai_response_background(
          self,
          query: str,
          conversation_id: str,
          user_id: str,
          node_ids: List[NodeContext],
          attachment_ids: Optional[List[str]] = None,
          run_id: str = None,
      ) -> AsyncGenerator[ChatMessageResponse, None]:
          """Background version for Celery tasks - reuses existing streaming logic"""
  
          async for chunk in self._generate_and_stream_ai_response(
              query, conversation_id, user_id, node_ids, attachment_ids
          ):
              yield chunk
  
      async def _prepare_attachments_as_images(
          self, attachment_ids: List[str]
      ) -> Optional[Dict[str, Dict[str, Union[str, int]]]]:
          """Convert attachment IDs directly to base64 images for multimodal processing"""
          try:
              if not attachment_ids:
                  return None
  
              images = {}
              for attachment_id in attachment_ids:
                  try:
                      # Get attachment info
                      attachment = await self.media_service.get_attachment(attachment_id)
                      logger.info(
                          f"DEBUG: Retrieved attachment {attachment_id}: type={attachment.attachment_type.value if attachment else 'None'}, mime_type={attachment.mime_type if attachment else 'None'}"
                      )
                      if (
                          attachment
                          and attachment.attachment_type.value.upper() == "IMAGE"
                      ):  # Check if it's an image
                          base64_data = await self.media_service.get_image_as_base64(
                              attachment_id
                          )
                          images[attachment_id] = {
                              "base64": base64_data,
                              "mime_type": attachment.mime_type,
                              "file_name": attachment.file_name,
                              "file_size": attachment.file_size,
                          }
                          logger.info(
                              f"Prepared image {attachment_id} ({attachment.file_name}) for multimodal processing"
                          )
                      else:
                          logger.info(
  