generation.py

"""
GenerationOrchestrator - RAG v2 Memecoin Generation Service

Manages the lifecycle of RAG v2 memecoin generation requests with integrated workflow
processing, session tracking, and background service management. Operates as an embedded
service eliminating the need for external workflow service.
"""

import asyncio
import traceback
import uuid
import os
from datetime import datetime
from typing import Optional, Dict, Any, List

from src.orchestrators.base_orchestrator.base_orchestrator import BaseOrchestrator
from src.services.ai.litellm_service import LiteLLMService
from src.services.ai.clip_embedding_service import CLIPEmbeddingService
from src.services.ai.image_generation_service import ImageGenerationService
from src.vector_store.memecoin_store import MemecoinVectorStore
from src.constants import CONFIG_ROOT, RES_ROOT
from src.util.config_loader import load_config
from src.models.llm.litellm_config import LiteLLMSetup
from src.domain.input_source.web_generation_input_source import WebGenerationInputSource
from src.domain.processor.rag_memecoin_generation_processor import (
    RAGMemecoinGenerationProcessor,
)
from src.domain.executor.executor_router import ExecutorRouter
from src.domain.workflow.default_workflow import DefaultWorkflow
from src.services.generated_meme_storage import (
    GeneratedMemeStorageService,
    LocalGeneratedMemeStorage,
    GeneratedMeme,
)


class GenerationOrchestrator(BaseOrchestrator):
    """
    Orchestrates RAG v2 memecoin generation with integrated workflow processing

    Features:
    - Embedded RAG workflow processing (no external service needed)
    - Real-time session tracking and status updates
    - Background service management with health monitoring
    - Service initialization and auto-recovery
    - Comprehensive error handling and logging
    """

    def __init__(self, edit_workflow=None, rag_workflow=None, **kwargs):
        """Initialize the generation orchestrator with all required services

        Args:
            edit_workflow: Optional pre-initialized GeneratedMemeEditAnalysisWorkflow (dependency injection)
            rag_workflow: Optional pre-initialized RAGMemecoinGenerationWorkflow (dependency injection)
            **kwargs: Optional service injections (currently for future extensibility)
        """
        super().__init__(name="GenerationOrchestrator")

        # Override start time to use int instead of datetime
        self._start_time: Optional[int] = None

        # Workflows (dependency injection pattern)
        self._edit_workflow = edit_workflow  # Pre-initialized GeneratedMemeEditAnalysisWorkflow
        self._rag_workflow = rag_workflow  # Pre-initialized RAGMemecoinGenerationWorkflow

        # Service instances (initialized lazily)
        self._llm_service: Optional[LiteLLMService] = None
        self._clip_service: Optional[CLIPEmbeddingService] = None
        self._image_service: Optional[ImageGenerationService] = None
        self._vector_store: Optional[MemecoinVectorStore] = None
        self._storage_service: Optional[GeneratedMemeStorageService] = None

        # Domain components (initialized after services)
        self._input_source: Optional[WebGenerationInputSource] = None
        self._processor: Optional[RAGMemecoinGenerationProcessor] = None
        self._executor_router: Optional[ExecutorRouter] = None
        self._workflow: Optional[DefaultWorkflow] = None

        # Session management
        self._session_tracking: Dict[str, Dict[str, Any]] = {}
        self._session_lock = asyncio.Lock()

        # Edit proposal management
        self._edit_proposals: Dict[str, tuple[datetime, Dict[str, Any]]] = {}
        self._proposals_lock = asyncio.Lock()
        self._proposal_ttl_seconds = 300  # 5 minutes

        # Background tasks (managed by base class)
        # Statistics
        self.processed_count = 0
        self.failed_count = 0
        self._last_generation_time: Optional[int] = None

        # Configuration
        self._session_cleanup_interval_seconds = 300  # 5 minutes
        self._max_session_age_seconds = 3600  # 1 hour

    async def _initialize_services(self) -> None:
        """
        Initialize services required by the orchestrator

        Implementation of BaseOrchestrator abstract method.
        """
        # Validate environment first
        await self._validate_environment()

        # Initialize all services
        await self._ensure_services_initialized()

        # Initialize domain components
        await self._initialize_domain_components()

    async def _run_background_loop(self) -> None:
        """
        Run background processing loop

        Implementation of BaseOrchestrator abstract method.
        Coordinates both session cleanup and workflow processing.
        """
        # Start workflow task
        workflow_task = asyncio.create_task(self._run_workflow_indefinitely())

        # Run session cleanup loop with workflow coordination
        cleanup_interval = 0
        while True:
            try:
                # Check workflow task health
                if workflow_task.done():
                    # Restart workflow if it failed
                    self.logger.warning("🔄 Workflow task died, restarting...")
                    workflow_task = asyncio.create_task(
                        self._run_workflow_indefinitely()
                    )

                # Periodic session cleanup
                cleanup_interval += 1
                if cleanup_interval >= (
                    self._session_cleanup_interval_seconds // 5
                ):  # Check every 5 seconds
                    await self._cleanup_old_sessions()
                    await self._cleanup_expired_proposals()
                    cleanup_interval = 0

                await asyncio.sleep(5)  # Check every 5 seconds

            except asyncio.CancelledError:
                # Cancel workflow task on shutdown
                if not workflow_task.done():
                    workflow_task.cancel()
                    try:
                        await workflow_task
                    except asyncio.CancelledError:
                        pass
                break
            except Exception as e:
                self.logger.error(f"❌ Background loop error: {e}")
                await asyncio.sleep(10)  # Wait before retry

    async def _check_health(self) -> Dict[str, Any]:
        """
        Check orchestrator-specific health status

        Implementation of BaseOrchestrator abstract method.

        Returns:
            Dictionary with health details specific to this orchestrator
        """
        # Check if core services are initialized and operational
        services_ready = all(
            [
                self._llm_service,
                self._clip_service,
                self._image_service,
                self._vector_store,
                self._storage_service,
                self._workflow,
            ]
        )

        # Check for session tracking issues
        sessions_count = len(self._session_tracking)
        sessions_healthy = sessions_count <= 100

        return {
            "services_ready": services_ready,
            "active_sessions": sessions_count,
            "sessions_healthy": sessions_healthy,
            "workflow_running": self._workflow is not None,
        }

    async def start_generation(self, request_data: Dict[str, Any]) -> Dict[str, Any]:
        """
        Start a new generation request

        Args:
            request_data: Generation request parameters

        Returns:
            Dict with session_id and initial status
        """
        if not self.is_running:
            raise RuntimeError("GenerationOrchestrator is not running")

        session_id = request_data.get("session_id") or str(uuid.uuid4())

        try:
            # Initialize session tracking
            async with self._session_lock:
                self._session_tracking[session_id] = {
                    "status": "pending",
                    "message": "Generation request accepted",
                    "progress": {"percentage": 0, "stage_description": "Initializing"},
                    "created_at": datetime.now().isoformat(),
                    "request_data": request_data,
                    "rag_examples_sent": False,  # Track if RAG examples were sent to client
                }

            # Start background processing (properly managed task)
            task = asyncio.create_task(
                self._process_generation_request(session_id, request_data)
            )
            # Store task reference to prevent garbage collection
            if not hasattr(self, "_active_generation_tasks"):
                self._active_generation_tasks = set()
            self._active_generation_tasks.add(task)
            task.add_done_callback(lambda t: self._active_generation_tasks.discard(t))

            self.logger.info(f"🎯 Started generation for session: {session_id[:8]}")

            return {"success": True, "session_id": session_id}

        except Exception as e:
            self.logger.error(f"❌ Failed to start generation: {e}")

            # Update session with error
            async with self._session_lock:
                if session_id in self._session_tracking:
                    self._session_tracking[session_id].update(
                        {
                            "status": "failed",
                            "message": f"Failed to start generation: {str(e)}",
                        }
                    )

            raise

    async def get_generation_status(self, session_id: str) -> Dict[str, Any]:
        """
        Get generation status for a session

        Args:
            session_id: Session identifier

        Returns:
            Dict with status, progress, and message
        """
        async with self._session_lock:
            if session_id not in self._session_tracking:
                return {"status": "not_found", "message": "Session not found"}

            return self._session_tracking[session_id].copy()

    async def mark_rag_examples_sent(self, session_id: str) -> None:
        """
        Mark RAG examples as sent to client for this session

        Args:
            session_id: Session identifier
        """
        async with self._session_lock:
            if session_id in self._session_tracking:
                self._session_tracking[session_id]["rag_examples_sent"] = True
                self.logger.debug(f"📤 Marked RAG examples as sent for session {session_id[:8]}")

    def _extract_generated_uuids(self, tokens: List[Dict[str, Any]]) -> List[str]:
        """
        Extract UUIDs from generated tokens

        Args:
            tokens: Raw tokens from backend

        Returns:
            List of UUID strings
        """
        uuids = []
        for token in tokens:
            uuid_val = token.get("uuid")
            if uuid_val:
                uuids.append(uuid_val)
            else:
                self.logger.warning("Token missing UUID, skipping")
        return uuids

    async def update_session_progress(
        self, session_id: str, stage_name: str, percentage: int, **additional_data
    ) -> None:
        """
        Update session progress with incremental workflow stage information

        This method is called by executors (via SessionStatusUpdateExecutor) to provide
        real-time progress updates as the generation workflow progresses through stages.
        Updates are stored in the orchestrator's runtime memory (_session_tracking) and
        are immediately available to the status endpoint.

        Thread-safe with async lock protection.

        Args:
            session_id: Session identifier to update
            stage_name: Name of the current workflow stage (e.g., "RAG Retrieval")
            percentage: Progress percentage (0-100)
            **additional_data: Stage-specific data (e.g., similar_examples_count, token_name)
        """
        async with self._session_lock:
            if session_id not in self._session_tracking:
                self.logger.warning(
                    f"⚠️  Session {session_id[:8]} not found for status update"
                )
                return

            # Apply random deviation to percentage (±5%) for visual variety
            import random

            deviation = random.randint(-5, 5)
            adjusted_percentage = max(0, min(100, percentage + deviation))

            # Update progress data
            self._session_tracking[session_id]["progress"] = {
                "percentage": adjusted_percentage,
                "stage_description": stage_name,
            }
            self._session_tracking[session_id]["current_stage"] = stage_name
            self._session_tracking[session_id]["status"] = (
                "completed" if adjusted_percentage >= 100 else "processing"
            )

            # Merge additional stage-specific data
            for key, value in additional_data.items():
                self._session_tracking[session_id][key] = value

            self.logger.info(
                f"📊 Session {session_id[:8]} progress: {stage_name} ({adjusted_percentage}%)"
            )

    async def get_orchestrator_status(self) -> Dict[str, Any]:
        """
        Get comprehensive orchestrator status

        Returns:
            Dict with orchestrator health and statistics
        """
        async with self._session_lock:
            active_sessions = len(
                [
                    s
                    for s in self._session_tracking.values()
                    if s["status"] in ["pending", "processing"]
                ]
            )

            return {
                "is_running": self.is_running,
                "is_healthy": self.is_running,
                "start_time": self._start_time,
                "active_sessions": active_sessions,
                "total_sessions": len(self._session_tracking),
                "processed_count": self.processed_count,
                "failed_count": self.failed_count,
                "last_generation_time": self._last_generation_time,
                "services": {
                    "llm_service": self._llm_service is not None,
                    "clip_service": self._clip_service is not None,
                    "image_service": self._image_service is not None,
                    "vector_store": self._vector_store is not None,
                    "storage_service": self._storage_service is not None,
                    "workflow": self._workflow is not None,
                },
            }

    async def check_health(self) -> Dict[str, Any]:
        """
        Health check endpoint replacement for workflow service

        Returns:
            Dict with health status and service information
        """
        try:
            status = await self.get_orchestrator_status()

            return {
                "healthy": self.is_running,
                "status": "ok" if self.is_running else "unhealthy",
                "orchestrator": status,
                "message": "GenerationOrchestrator is operational"
                if self.is_running
                else "GenerationOrchestrator needs attention",
            }

        except Exception as e:
            return {
                "healthy": False,
                "status": "error",
                "message": f"Health check failed: {str(e)}",
            }

    @property
    def storage_service(self) -> Optional[GeneratedMemeStorageService]:
        """
        Get the storage service instance

        Returns:
            Storage service if initialized, None otherwise
        """
        return self._storage_service

    async def _validate_environment(self) -> None:
        """Validate required environment variables and configuration"""
        self.logger.info("🔍 Validating environment configuration...")

        required_keys = {
            "REPLICATE_API_KEY": "CLIP embedding service",
            "OPENAI_API_KEY": "LLM services (primary)",
        }

        optional_keys = {
            "OPENROUTER_API_KEY": "OpenRouter LLM services (fallback)",
            "LANGFUSE_SECRET_KEY": "LLM observability (optional)",
        }

        missing_required = []
        missing_optional = []

        # Check required keys
        for key, service in required_keys.items():
            if not os.getenv(key):
                missing_required.append(f"{key} ({service})")

        # Check optional keys
        for key, service in optional_keys.items():
            if not os.getenv(key):
                missing_optional.append(f"{key} ({service})")

        # Report missing keys
        if missing_required:
            error_msg = (
                f"Missing required environment variables: {', '.join(missing_required)}"
            )
            self.logger.error(f"❌ {error_msg}")
            raise RuntimeError(error_msg)

        if missing_optional:
            self.logger.warning(
                f"⚠️ Missing optional environment variables: {', '.join(missing_optional)}"
            )

        self.logger.info("✅ Environment validation completed")

    async def _ensure_services_initialized(self) -> None:
        """Initialize all required services"""
        self.logger.info("🔧 Initializing services...")

        try:
            # Initialize LLM service
            if not self._llm_service:
                config_path = CONFIG_ROOT / "litellm.yaml"
                config = load_config(LiteLLMSetup, str(config_path))
                self._llm_service = LiteLLMService(config=config)
                await self._llm_service.initialize()

            # Initialize CLIP service
            if not self._clip_service:
                self._clip_service = CLIPEmbeddingService()
                await self._clip_service.initialize()

            # Initialize image generation service
            if not self._image_service:
                self._image_service = ImageGenerationService()

            # Initialize vector store
            if not self._vector_store:
                db_path = RES_ROOT / "memecoins" / "rag_db"
                self._vector_store = MemecoinVectorStore(str(db_path))

            # Initialize storage service
            if not self._storage_service:
                self._storage_service = LocalGeneratedMemeStorage()
                await self._storage_service.initialize()

            self.logger.info("✅ All services initialized successfully")

        except Exception as e:
            self.logger.error(f"❌ Service initialization failed: {e}")
            raise

    async def _initialize_domain_components(self) -> None:
        """Initialize domain architecture components via RAGMemecoinGenerationWorkflow"""
        self.logger.info("🏗️ Initializing domain components via workflow...")

        try:
            # Use pre-initialized RAG workflow (dependency injection)
            if not self._rag_workflow:
                raise RuntimeError(
                    "RAGMemecoinGenerationWorkflow not provided during initialization"
                )

            # Initialize workflow with orchestrator services (orchestrator mode)
            # Note: The workflow may already be initialized by the registry,
            # but we need to extract component references

            # Extract component references for direct access
            # (maintains compatibility with existing orchestrator code)
            self._workflow = self._rag_workflow.workflow
            self._input_source = self._rag_workflow.input_source
            self._processor = self._rag_workflow.processor
            self._executor_router = self._rag_workflow.executor

            # Inject orchestrator reference into executor router for callback pattern
            # This enables executors to update session progress in real-time
            self._executor_router.set_orchestrator(self)

            # Initialize the workflow to register event callbacks
            # This must be done before accepting requests
            if not self._workflow.running:
                await self._workflow.initialize()
                self.logger.info("✅ Workflow initialized with event callbacks")

            self.logger.info("✅ Domain components initialized successfully")

        except Exception as e:
            self.logger.error(f"❌ Domain component initialization failed: {e}: {traceback.format_exc()}")
            raise

    # Removed _start_background_tasks - now handled by base class

    async def _run_workflow_indefinitely(self) -> None:
        """Run the workflow indefinitely to process incoming requests"""
        restart_count = 0
        max_restarts = 3

        while restart_count < max_restarts:
            try:
                self.logger.info("🎨 Starting workflow infinite processing loop...")
                await self._workflow.run_indefinitely()
                break  # If we get here, workflow ended normally
            except asyncio.CancelledError:
                self.logger.info("⏹️ Workflow infinite loop cancelled")
                break
            except Exception as e:
                restart_count += 1
                self.logger.error(
                    f"❌ Workflow infinite loop error (attempt {restart_count}/{max_restarts}): {e}"
                )

                if restart_count < max_restarts and self.is_running:
                    self.logger.info(
                        f"🔄 Attempting to restart workflow (attempt {restart_count + 1}/{max_restarts})..."
                    )
                    await asyncio.sleep(5 * restart_count)  # Exponential backoff
                else:
                    self.logger.error(
                        "❌ Max restart attempts reached, workflow will not restart"
                    )
                    break

    # Removed _session_cleanup_loop - now integrated into _run_background_loop

    async def _cleanup_old_sessions(self) -> None:
        """Remove old sessions from tracking"""
        current_time = datetime.now().timestamp()

        async with self._session_lock:
            to_remove = []

            for session_id, session_data in self._session_tracking.items():
                created_at = datetime.fromisoformat(session_data["created_at"])
                age = current_time - created_at.timestamp()

                if age > self._max_session_age_seconds:
                    to_remove.append(session_id)

            for session_id in to_remove:
                del self._session_tracking[session_id]

            if to_remove:
                self.logger.info(f"🧹 Cleaned up {len(to_remove)} old sessions")

    async def _cleanup_expired_proposals(self) -> None:
        """Remove expired edit proposals from cache"""
        current_time = datetime.now()

        async with self._proposals_lock:
            to_remove = []

            for proposal_uuid, (timestamp, _) in self._edit_proposals.items():
                age_seconds = (current_time - timestamp).total_seconds()

                if age_seconds > self._proposal_ttl_seconds:
                    to_remove.append(proposal_uuid)

            for proposal_uuid in to_remove:
                del self._edit_proposals[proposal_uuid]

            if to_remove:
                self.logger.info(f"🧹 Cleaned up {len(to_remove)} expired edit proposals")

    async def generate_generated_edit_proposal(
        self,
        uuid: str,
        user_feedback: str,
        current_metadata: Optional[Dict[str, Any]] = None,
    ) -> tuple[bool, str, Optional[Dict[str, Any]]]:
        """
        Generate edit proposal for generated memecoin using intelligent workflow

        Args:
            uuid: Generated memecoin UUID
            user_feedback: User's free-form feedback about what to change
            current_metadata: Optional current token metadata from UI (if user edited fields).
                             Contains: {name, ticker, description, tags}.
                             If None, loads from storage. If provided, uses as baseline.

        Returns:
            Tuple of (success, message, result_dict)
            result_dict contains: {"proposal_uuid": str, "proposal": dict}
        """
        try:
            self.logger.info(f"🔧 Generating edit proposal for: {uuid[:8]}")
            self.logger.info(f"💬 Feedback: {user_feedback[:100]}...")
            self.logger.info(f"🔑 Using UUID as session_id for proposal caching: {uuid[:8]}")

            # Load memecoin from storage for image and generation_info
            stored_memecoin = await self._storage_service.load(uuid)
            if not stored_memecoin:
                return False, f"Memecoin {uuid} not found", None

            # Use provided metadata if available, otherwise use stored data
            if current_metadata:
                self.logger.info("📝 Using current metadata from UI (potentially user-edited)")

                # Log what user changed (helpful for debugging)
                stored_name = stored_memecoin.name
                stored_ticker = stored_memecoin.ticker
                stored_description = stored_memecoin.description

                current_name = current_metadata.get("name", stored_name)
                current_ticker = current_metadata.get("ticker", stored_ticker)
                current_description = current_metadata.get("description", stored_description)

                if stored_name != current_name:
                    self.logger.info(f"   Name changed: {stored_name} → {current_name}")
                if stored_ticker != current_ticker:
                    self.logger.info(f"   Ticker changed: {stored_ticker} → {current_ticker}")
                if stored_description != current_description:
                    desc_len_old = len(stored_description)
                    desc_len_new = len(current_description)
                    self.logger.info(f"   Description changed (length: {desc_len_old} → {desc_len_new})")

                token_name = current_name
                ticker = current_ticker
                description = current_description
                tags = current_metadata.get("tags", getattr(stored_memecoin, 'tags', []))
            else:
                self.logger.info("📝 Using stored metadata (no user edits detected)")
                token_name = stored_memecoin.name
                ticker = stored_memecoin.ticker
                description = stored_memecoin.description
                tags = getattr(stored_memecoin, 'tags', [])

            # Convert to dict for workflow
            memecoin_data = {
                "uuid": uuid,
                "session_id": uuid,  # Use UUID as session_id for proposal caching
                "token_name": token_name,
                "ticker": ticker,
                "description": description,
                "tags": tags,
                "image_base64": stored_memecoin.image_data,  # Always from storage
                "image_mime_type": "image/png",
                "generation_info": stored_memecoin.generation_info,  # Always from storage
                "created_at": 0,  # Not used in editing
            }

            # Use pre-initialized edit analysis workflow (dependency injection)
            if not self._edit_workflow:
                return (
                    False,
                    "Edit workflow not initialized - orchestrator configuration error",
                    None,
                )

            # Run workflow to get edited memecoin (workflow is pre-initialized)
            edited_entry = await self._edit_workflow.run(memecoin_data, user_feedback)

            # Generate proposal UUID
            import uuid as uuid_module

            proposal_uuid = str(uuid_module.uuid4())

            # Get edit summary for metadata
            edit_summary = self._edit_workflow.get_edit_summary()

            # Format image_data with proper data URL prefix
            image_data = None
            if edit_summary.get("image_regenerated") and edited_entry.image_base64:
                # Check if already has data URL prefix
                if edited_entry.image_base64.startswith("data:"):
                    image_data = edited_entry.image_base64
                else:
                    # Add data URL prefix for raw Base64
                    image_data = f"data:image/png;base64,{edited_entry.image_base64}"

            # Build proposal data
            proposal_data = {
                "name": edited_entry.token_name,
                "ticker": edited_entry.ticker,
                "description": edited_entry.description,
                "tags": edited_entry.tags,
                "image_data": image_data,
                "regenerated_image": edit_summary.get("image_regenerated", False),
                "edit_summary": edit_summary,
            }

            # Cache proposal with timestamp
            async with self._proposals_lock:
                self._edit_proposals[proposal_uuid] = (datetime.now(), proposal_data)

            self.logger.info(f"✅ Edit proposal generated: {proposal_uuid[:8]}")
            self.logger.info(f"📊 Strategy: {edit_summary.get('strategy')}")
            self.logger.info(f"🖼️ Image regenerated: {edit_summary.get('image_regenerated')}")

            return True, "Proposal generated successfully", {
                "proposal_uuid": proposal_uuid,
                "proposal": proposal_data,
            }

        except Exception as e:
            self.logger.error(f"❌ Failed to generate edit proposal: {e}")
            return False, f"Failed to generate proposal: {str(e)}", None

    async def accept_generated_edit_proposal(
        self,
        proposal_uuid: str,
        uuid: str,
        edited_proposal: Optional[Dict[str, Any]] = None,
    ) -> tuple[bool, str]:
        """
        Accept and apply edit proposal to generated memecoin

        Args:
            proposal_uuid: UUID of cached proposal
            uuid: Generated memecoin UUID
            edited_proposal: Optional manually edited proposal (overrides cached)

        Returns:
            Tuple of (success, message)
        """
        try:
            self.logger.info(f"📥 Accepting edit proposal: {proposal_uuid[:8]} for {uuid[:8]}")

            # Retrieve proposal (cached or provided)
            if edited_proposal:
                proposal_data = edited_proposal
                self.logger.info("Using manually edited proposal")
            else:
                # Get from cache with TTL check
                async with self._proposals_lock:
                    if proposal_uuid not in self._edit_proposals:
                        return False, "Proposal not found or expired"

                    timestamp, proposal_data = self._edit_proposals[proposal_uuid]

                    # Check TTL
                    age_seconds = (datetime.now() - timestamp).total_seconds()
                    if age_seconds > self._proposal_ttl_seconds:
                        del self._edit_proposals[proposal_uuid]
                        return False, "Proposal expired (5 minute limit)"

            # Load existing memecoin
            existing = await self._storage_service.load(uuid)
            if not existing:
                return False, f"Memecoin {uuid} not found"

            # Apply updates (in-place, same UUID)
            existing.name = proposal_data.get("name", existing.name)
            existing.ticker = proposal_data.get("ticker", existing.ticker)
            existing.description = proposal_data.get("description", existing.description)

            # Update image if regenerated
            if proposal_data.get("image_data"):
                existing.image_data = proposal_data["image_data"]

            # Save updated memecoin (same UUID)
            success = await self._storage_service.save(existing)

            if success:
                # Remove proposal from cache
                if not edited_proposal:
                    async with self._proposals_lock:
                        if proposal_uuid in self._edit_proposals:
                            del self._edit_proposals[proposal_uuid]

                self.logger.info(f"✅ Edit applied successfully to {uuid[:8]}")
                return True, "Changes applied successfully"

            return False, "Failed to save changes to storage"

        except Exception as e:
            self.logger.error(f"❌ Failed to accept edit proposal: {e}")
            return False, f"Failed to apply changes: {str(e)}"

    async def _process_generation_request(
        self, session_id: str, request_data: Dict[str, Any]
    ) -> None:
        """
        Background task to process generation request using domain workflow

        Args:
            session_id: Session identifier
            request_data: Generation request parameters
        """
        try:
            # Update status to processing
            async with self._session_lock:
                self._session_tracking[session_id].update(
                    {
                        "status": "processing",
                        "message": "Processing generation request",
                        "progress": {
                            "percentage": 10,
                            "stage_description": "Starting workflow",
                        },
                    }
                )

            # Process through domain workflow
            # This will be implemented in the next step
            await self._run_workflow_processing(session_id, request_data)

        except Exception as e:
            self.logger.error(
                f"❌ Generation processing failed for {session_id[:8]}: {e}"
            )
            self.failed_count += 1

            # Update session with failure
            async with self._session_lock:
                if session_id in self._session_tracking:
                    self._session_tracking[session_id].update(
                        {"status": "failed", "message": f"Generation failed: {str(e)}"}
                    )

    async def _run_workflow_processing(
        self, session_id: str, request_data: Dict[str, Any]
    ) -> None:
        """
        Run the actual workflow processing using domain architecture

        Args:
            session_id: Session identifier for tracking
            request_data: Generation request parameters
        """
        try:
            self.logger.info(
                f"🎨 Starting workflow processing for session: {session_id[:8]}"
            )

            # Process the request through input source (triggers domain workflow)
            success = await self._input_source.process_request(request_data)

            if not success:
                raise RuntimeError("Workflow processing failed")

        except Exception as e:
            self.logger.error(
                f"❌ Workflow processing failed for {session_id[:8]}: {e}"
            )

            # Update session with failure
            async with self._session_lock:
                if session_id in self._session_tracking:
                    self._session_tracking[session_id].update(
                        {
                            "status": "failed",
                            "message": f"Generation failed: {str(e)}",
                            "progress": {
                                "percentage": 0,
                                "stage_description": "Failed",
                            },
                        }
                    )

            raise

    async def _update_session_progress(
        self, session_id: str, stage_description: str, percentage: int
    ) -> None:
        """Update session progress in tracking"""
        async with self._session_lock:
            if session_id in self._session_tracking:
                self._session_tracking[session_id]["progress"] = {
                    "percentage": percentage,
                    "stage_description": stage_description,
                }
                self._session_tracking[session_id]["message"] = (
                    f"Processing: {stage_description}"
                )

    async def _extract_results_from_action(self, action) -> Dict[str, Any]:
        """
        Extract results from workflow action for API response

        Args:
            action: Action object from workflow processing

        Returns:
            Dict with formatted results for API response
        """
        try:
            # Extract generated tokens from action
            generated_tokens = []
            storage_uuids = []

            # Check if action has generated tokens
            if hasattr(action, "data") and isinstance(action.data, dict):
                tokens = action.data.get("generated_tokens", [])
                for token in tokens:
                    if isinstance(token, dict):
                        # Save each generated token to storage
                        memecoin_uuid = str(uuid.uuid4())

                        # Create GeneratedMemecoin object
                        memecoin = GeneratedMeme(
                            uuid=memecoin_uuid,
                            name=token.get("name", ""),
                            ticker=token.get("ticker", ""),
                            description=token.get("description", ""),
                            image_data=token.get("image_data_uri", ""),
                            tags=token.get("tags", []),
                            generation_info={
                                "uuid": memecoin_uuid,
                                "session_id": action.data.get("session_id", ""),
                                "timestamp": datetime.now().isoformat(),
                                "generation_type": "rag_v2_generation",
                                "prompt": action.data.get("prompt", ""),
                                "rag_examples": action.data.get("rag_examples", []),
                            },
                            created_at=datetime.now().isoformat(),
                        )

                        # Save to storage
                        if self._storage_service:
                            save_success = await self._storage_service.save(memecoin)
                            if save_success:
                                storage_uuids.append(memecoin_uuid)
                                self.logger.info(
                                    f"💾 Saved generated memecoin to storage: {memecoin_uuid[:8]}"
                                )
                            else:
                                self.logger.warning(
                                    f"⚠️ Failed to save memecoin to storage: {memecoin_uuid[:8]}"
                                )

                        generated_tokens.append(token)

            return {
                "success": True,
                "message": "Generation completed successfully",
                "generated_tokens": generated_tokens,
                "storage_uuids": storage_uuids,  # UUIDs of saved memecoins
                "rag_examples": action.data.get("rag_examples", [])
                if hasattr(action, "data")
                else [],
                "metadata": {
                    "saved_count": len(storage_uuids),
                    "total_generated": len(generated_tokens),
                },
            }

        except Exception as e:
            self.logger.error(f"❌ Error extracting results: {e}")
            return {"success": False, "message": f"Error processing results: {str(e)}"}

    async def _cleanup_resources(self) -> None:
        """
        Cleanup orchestrator-specific resources

        Implementation of BaseOrchestrator abstract method.
        """
        # Cancel any active generation tasks
        if hasattr(self, "_active_generation_tasks"):
            for task in list(self._active_generation_tasks):
                if not task.done():
                    task.cancel()
                    try:
                        await task
                    except asyncio.CancelledError:
                        pass
            self._active_generation_tasks.clear()

        # Reset services (they can be re-initialized later)
        self._llm_service = None
        self._clip_service = None
        self._image_service = None
        self._vector_store = None
        self._storage_service = None
        self._input_source = None
        self._processor = None
        self._executor_router = None
        self._workflow = None

        # Clear session tracking
        async with self._session_lock:
            self._session_tracking.clear()

        self.logger.info("🧹 GenerationOrchestrator cleanup completed")