import inspect
import importlib
from pathlib import Path
import functools
import json
import html
import os
import re
from dotenv import load_dotenv, set_key
import asyncio
import aiohttp
from datetime import datetime
from fasthtml.common import *
from fastlite import Database  # CORRECTED: Use the Database class
from loguru import logger
import imports.server_logging as slog
import config as CFG
from config import COLOR_MAP
from imports import botify_code_generation
from imports.stream_orchestrator import stream_orchestrator
from typing import AsyncGenerator, Optional
import imports.server_logging as slog

log = slog.LogManager(logger)

import logging

# Silence the piper logger
logging.getLogger("piper").setLevel(logging.ERROR)

# Strong references for background asyncio tasks to prevent ipykernel garbage collection warnings
_background_tasks = set()


def fire_and_forget(coro):
    """Safely executes an async task in the background without Jupyter garbage collecting it."""
    task = asyncio.create_task(coro)
    _background_tasks.add(task)
    task.add_done_callback(_background_tasks.discard)
    return task


def title_name(word: str) -> str:
    """Format a string into a title case form."""
    if not word:
        return ''
    formatted = word.replace('.', ' ').replace('-', ' ')
    words = []
    for part in formatted.split('_'):
        words.extend(part.split())
    processed_words = []
    for word in words:
        if word.isdigit():
            processed_words.append(word.lstrip('0') or '0')
        else:
            processed_words.append(word.capitalize())
    return ' '.join(processed_words)


def pipeline_operation(func):

    @functools.wraps(func)
    def wrapper(self, *args, **kwargs):
        url = args[0] if args else None
        if not url:
            return func(self, *args, **kwargs)
        old_state = self._get_clean_state(url)
        result = func(self, *args, **kwargs)
        new_state = self._get_clean_state(url)
        if old_state != new_state:
            changes = {k: new_state[k] for k in new_state if k not in old_state or old_state[k] != new_state[k]}
            if changes:
                operation = func.__name__
                step_changes = [k for k in changes if not k.startswith('_')]
                if step_changes:
                    log.pipeline(f"Operation '{operation}' updated state", details=f"Steps: {', '.join(step_changes)}", pipeline_id=url)
                # Use Rich JSON display for pipeline changes
                formatted_changes = slog.rich_json_display(changes, console_output=False, log_output=True)
                log.debug('pipeline', f"Pipeline '{url}' detailed changes", formatted_changes)
        return result
    return wrapper


def db_operation(func):

    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        try:
            result = func(*args, **kwargs)
            if func.__name__ == '__setitem__':
                key, value = (args[1], args[2])
                if not key.startswith('_') and (not key.endswith('_temp')):
                    if key in ('last_app_choice', 'last_profile_id', 'last_visited_url', 'pipeline_id'):
                        log.data(f'State updated: {key}', value)
                    else:
                        log.debug('database', f'DB {func.__name__}: {key}', f'value: {str(value)[:30]}...' if len(str(value)) > 30 else f'value: {value}')
            return result
        except Exception as e:
            # Don't log KeyError as ERROR for __getitem__ - it's expected behavior
            if func.__name__ == '__getitem__' and isinstance(e, KeyError):
                logger.debug(f'Key not found in database: {e}')
            else:
                log.error(f'Database operation {func.__name__} failed', e)
            raise
    return wrapper


class DictLikeDB:

    def __init__(self, store, Store):
        self.store = store
        logger.debug('DictLikeDB initialized.')

    @db_operation
    def __getitem__(self, key):
        try:
            record = self.store[key]
            # 🎯 UNIFIED FIX: Handle both dicts (notebook) and objects (server)
            value = record['value'] if isinstance(record, dict) else record.value
            logger.debug(f'Retrieved from DB: {key} = {value}')
            return value
        except NotFoundError:
            # Don't log as error - this is expected behavior when checking for keys
            logger.debug(f'Key not found: {key}')
            raise KeyError(key)

    @db_operation
    def __setitem__(self, key, value):
        # Explicitly declare pk='key' to prevent FastLite from defaulting to 'rowid'
        self.store.upsert({'key': key, 'value': value}, pk='key')
        logger.debug(f'Saved to persistence store: {key} = {value}')

    @db_operation
    def __delitem__(self, key):
        try:
            self.store.delete(key)
            if key != 'temp_message':
                logger.warning(f'Deleted key from persistence store: {key}')
        except NotFoundError:
            logger.error(f'Attempted to delete non-existent key: {key}')
            raise KeyError(key)

    @db_operation
    def __contains__(self, key):
        exists = key in self.store
        logger.debug(f"Key '<{COLOR_MAP['key']}>{key}</{COLOR_MAP['key']}>' exists: <{COLOR_MAP['value']}>{exists}</{COLOR_MAP['value']}>")
        return exists

    @db_operation
    def __iter__(self):
        for record in self.store():
            # 🎯 UNIFIED FIX
            yield record['key'] if isinstance(record, dict) else record.key

    @db_operation
    def items(self):
        for record in self.store():
            # 🎯 UNIFIED FIX
            k = record['key'] if isinstance(record, dict) else record.key
            v = record['value'] if isinstance(record, dict) else record.value
            yield (k, v)

    @db_operation
    def keys(self):
        return list(self)

    @db_operation
    def values(self):
        for record in self.store():
            # 🎯 UNIFIED FIX
            yield record['value'] if isinstance(record, dict) else record.value

    @db_operation
    def get(self, key, default=None):
        try:
            return self[key]
        except KeyError:
            logger.debug(f"Key '<{COLOR_MAP['key']}>{key}</{COLOR_MAP['key']}>' not found. Returning default: <{COLOR_MAP['value']}>{default}</{COLOR_MAP['value']}>")
            return default

    @db_operation
    def set(self, key, value):
        self[key] = value
        return value


class Pipulate:
    """Central coordinator for pipelines and chat functionality.

    This class serves as the main interface for plugins to access
    shared functionality without relying on globals.

    As Pipulate evolves toward its "Digital Workshop" vision, this coordinator
    will support:
    - Sub-plugin architecture (steps expanding to full-screen apps)
    - Content curation systems (archive surfing, variant creation)
    - Progressive distillation workflows (search, sort, sieve, story)
    - Local-first creative exploration with privacy preservation

    The centralized coordination pattern enables sophisticated interaction
    monitoring and behavioral pattern analysis while maintaining the
    "vibrating edge" of creative freedom that powers genuine innovation.
    """
    PRESERVE_REFILL = True
    UNLOCK_BUTTON_LABEL = '🔓 Unlock'

    # START: pipulate_init
    def __init__(self, pipeline_table=None, db=None, friendly_names=None, append_func=None, get_profile_id_func=None, get_profile_name_func=None, chat_instance=None, db_path=None):
        self.chat = chat_instance
        self.friendly_names = friendly_names
        self.append_to_conversation = append_func
        self.get_current_profile_id = get_profile_id_func
        self.get_profile_name = get_profile_name_func

        self.active_local_model = CFG.DEFAULT_PROMPT_MODEL
        self.active_cloud_model = None
        self.message_queue = self.OrderedMessageQueue()
        self.is_notebook_context = bool(db_path) # Flag for notebook context
        self.dialogue_tree = {} # Container for centralized narrative scripts

        # 🌌 THE TOPOLOGICAL MANIFOLD (Unified Path Registry)
        actual_root = self._find_project_root(Path(__file__).resolve()) or Path.cwd()
        
        class WorkspaceManifold:
            def __init__(self, root: Path, is_notebook: bool):
                self.root = root
                self.base = root / "Notebooks" if is_notebook else root
                self.data = self.base / "data"
                self.logs = self.data / "logs"
                self.temp = self.data / "temp"
                self.downloads = self.data / "downloads"
                self.browser_cache = self.base / "browser_cache"
                self.deliverables = self.base / "Deliverables"
                
                # Force the physical reality into existence
                for p in [self.data, self.logs, self.temp, self.downloads, self.browser_cache, self.deliverables]:
                    p.mkdir(parents=True, exist_ok=True)
                    
        self.paths = WorkspaceManifold(actual_root, self.is_notebook_context)

        if db_path:
            # Standalone/Notebook Context: Create our "Parallel Universe" DB using fastlite directly
            from fastlite import Database
            from loguru import logger
            logger.info(f"Pipulate initializing in standalone mode with db: {db_path}")
    
            # 1. Create a database connection using fastlite.Database
            db_conn = Database(db_path)
    
            # 2. Access the table handles via the .t property
            l_store = db_conn.t.store
            l_pipeline = db_conn.t.pipeline
            # Note: We don't need to explicitly create tables; fastlite handles it.
    
            self.pipeline_table = l_pipeline
            # The second argument `Store` from fast_app isn't needed by DictLikeDB.
            self.db = DictLikeDB(l_store, None)
    
            # In standalone mode, some features that rely on the server are stubbed out
            if self.append_to_conversation is None: self.append_to_conversation = lambda msg, role: print(f"[{role}] {msg}")
            if self.get_current_profile_id is None: self.get_current_profile_id = lambda: 'standalone'
            if self.get_profile_name is None: self.get_profile_name = lambda: 'standalone'
    
        else:
            # Server Context: Use the objects passed in from server.py
            from loguru import logger
            logger.info("Pipulate initializing in server mode.")
            self.pipeline_table = pipeline_table
            self.db = db
    # END: pipulate_init

    def negotiate_ai_models(self, preferred_local: str = None, preferred_cloud: str = None) -> dict:
        """
        Uses the Universal Adapter (llm) to verify AI readiness using fuzzy matching
        against a prioritized list of preferred models. Elevates model selection to a global capability.
        """
        import llm
        
        if preferred_local:
            print("Scanning for your preferred local models...")
        else:
            print("Scanning your system for available AI models...")

        try:
            # 1. Gather all models known to the Universal Adapter
            available_models = [m.model_id for m in llm.get_models()]
            
            # 2. Check for ANY local model (Ollama models typically lack provider prefixes)
            has_local = any('ollama' in str(type(m)).lower() for m in llm.get_models())

            # 3. Process User Preferences
            def parse_preferences(pref_string):
                if not pref_string: return []
                return [p.strip().lower() for p in pref_string.split(',')]

            local_prefs = parse_preferences(preferred_local)
            cloud_prefs = parse_preferences(preferred_cloud)

            selected_local = None
            selected_cloud = None

            # 4. Fuzzy Matching Logic (Find highest priority match)
            for pref in local_prefs:
                match = next((m for m in available_models if pref in m.lower() and 'ollama' in str(type(llm.get_model(m))).lower()), None)
                if match:
                    selected_local = match
                    break 

            for pref in cloud_prefs:
                match = next((m for m in available_models if pref in m.lower() and 'ollama' not in str(type(llm.get_model(m))).lower()), None)
                if match:
                    selected_cloud = match
                    break 

            # 5. Reporting and Graceful Degradation
            if selected_local:
                print(f"Excellent. Local model '{selected_local}' is active and ready.")
                print(f"\n✅ Locked in Local Model: {selected_local}")
                self.db['active_local_model'] = selected_local
            elif has_local:
                print("I found local models, but not your preferred choices.")
                print(f"\nℹ️  Preferred local models not found, but other local models are available.")
                print(f"Available models: {', '.join([m for m in available_models if 'ollama' in str(type(llm.get_model(m))).lower()])}")
                selected_local = True 
            else:
                print("I do not detect a local AI brain on your system.")
                print("\nℹ️  Ollama is not running or not installed.")
                print("Pipulate works perfectly fine without it, but a local AI 'riding shotgun' ensures privacy.")
                print("\nTo upgrade your environment for true Local-First Sovereignty:")
                print("1. Go to https://ollama.com/")
                print("2. Download the installer for your host operating system.")
                print(f"3. Install it, open a terminal, run 'ollama run {CFG.DEFAULT_PROMPT_MODEL.split(':')[0]}', and try again.")
                
            if selected_cloud:
                 print(f"✅ Locked in Cloud Model: {selected_cloud}")

            return {
                "local": selected_local,
                "cloud": selected_cloud,
                "has_any_local": has_local
            }

        except Exception as e:
            print(f"❌ Error communicating with the Universal Adapter: {e}")
            return {"local": False, "cloud": False, "has_any_local": False}

    def get_home_menu_item(self) -> str:
        """Returns the appropriate home menu item text based on the HOME_APP setting."""
        home_app_name = getattr(CFG, 'HOME_APP', '030_roles') # Default to '030_roles'
        return self.friendly_names.get(home_app_name, title_name(home_app_name))

    def endpoint_name(self, endpoint: str) -> str:
        if not endpoint:
            return self.get_home_menu_item()
        if endpoint in self.friendly_names:
            return self.friendly_names[endpoint]
        return title_name(endpoint)

    def append_to_conversation_from_instance(self, message: str, role: str = 'user'):
        """Instance method wrapper for the global self.append_to_conversation function."""
        return self.append_to_conversation(message, role=role)

    def append_to_history(self, message: str, role: str = 'system') -> None:
        """Add a message to the LLM conversation history without triggering a response.

        This is the preferred way for workflows to update the LLM's context about:
        - UI state changes
        - Form submissions
        - Validation results
        - Explanatory text shown to users
        - Step completion status

        Args:
            message: The message to add to history
            role: The role of the message sender ("system", "user", "assistant")
        """
        self.append_to_conversation(message, role=role)

    class OrderedMessageQueue:
        """A lightweight queue to ensure messages are delivered in order.

        This class creates a simple message queue that ensures messages are delivered
        in the exact order they are added, without requiring explicit delays between
        messages. It's used to fix the message streaming order issues.

        As part of the Digital Workshop evolution, this queue will support:
        - Interaction pattern recognition for adaptive workflows
        - State transition tracking for sub-plugin applications
        - Creative session analysis for distillation workflows
        - Privacy-preserving behavioral insights for local optimization

        The workflow state tracking enables sophisticated user interaction
        analysis while keeping all data local for maximum creative freedom.
        """

        def __init__(self):
            self.queue = []
            self._processing = False
            self._current_step = None
            self._step_started = False
            self._step_complete = False
            self._workflow_context = None

        async def add(self, pipulate, message, **kwargs):
            """Add a message to the queue and process if not already processing.

            This method no longer handles conversation history - that's now managed by pipulate.stream.
            """
            logger.info(f'[🔄 QUEUEING] {message[:100]}...')
            self.queue.append((pipulate, message, kwargs))
            if not self._processing:
                await self._process_queue()

        async def _process_queue(self):
            """Process messages in the queue.

            This method now focuses solely on queue processing and streaming,
            leaving conversation history management to pipulate.stream.
            """
            if self._processing:
                return

            self._processing = True
            try:
                while self.queue:
                    pipulate, message, kwargs = self.queue.pop(0)
                    await pipulate.stream(message, **kwargs)
            finally:
                self._processing = False

        def mark_step_complete(self, step_num):
            """Mark a step as completed."""
            self._current_step = step_num
            self._step_complete = True

        def mark_step_started(self, step_num):
            """Mark a step as started but not completed."""
            self._current_step = step_num
            self._step_started = True

    def open_folder(self, path_str: str = "."):
        """Opens the specified folder in the host system's default file explorer."""
        import platform
        import subprocess
        from pathlib import Path
        
        folder_path = Path(path_str).resolve()
        
        if not folder_path.exists() or not folder_path.is_dir():
            self.logger.error(f"❌ Error: Path is not a valid directory: {folder_path}")
            return False

        system = platform.system()
        try:
            if system == "Windows":
                os.startfile(folder_path)
            elif system == "Darwin":  # macOS
                subprocess.run(["open", folder_path])
            else:  # Linux (xdg-open covers most desktop environments)
                subprocess.run(["xdg-open", folder_path])
            return True
        except Exception as e:
            self.logger.error(f"❌ Failed to open folder. Error: {e}")
            return False

    def speak(self, text: str, delay: float = 0.0, wait: bool = True, emoji: str = None):
        """
        Synthesizes text to speech using the global ChipVoiceSystem if available.
        Fails gracefully to simple printing if the audio backend is unavailable.
        Automatically routes Markdown links to the UI as tightly-spaced HTML, 
        while stripping them for the TTS engine.
        Bracketed text like [🏆] is shown visually but omitted from speech.
        """
        import re
        
        display_emoji = emoji if emoji is not None else CFG.WAND_SPEAKS_EMOJI

        # 1. The Acoustic Payload 
        # Extract link text: "[weird stuff](url)" -> "weird stuff"
        voice_text = re.sub(r'\[([^\]]+)\]\([^\)]+\)', r'\1', text)
        # Remove silent tags entirely: "active [🏆]" -> "active "
        # Strip HTML tags so the TTS doesn't read them aloud
        voice_text = re.sub(r'<[^>]+>', '', voice_text)
        voice_text = re.sub(r'\[[^\]]+\]', '', voice_text)

        # 2. The Visual Payload (Hidden IPython complexity)
        if getattr(self, 'is_notebook_context', False):
            try:
                from IPython.display import display, HTML
                
                # Parse the Markdown link into an HTML anchor tag
                html_text = re.sub(r'\[([^\]]+)\]\(([^\)]+)\)', r'<a href="\2" target="_blank">\1</a>', text)
                
                # Strip the brackets from silent tags so they display normally
                html_text = re.sub(r'\[([^\]]+)\]', r'\1', html_text)
                
                # Convert newlines to HTML breaks to perfectly mimic print() spacing
                html_text = html_text.replace('\n', '<br>')
                
                # Render as a tightly packed block with zero bottom margin
                display(HTML(f'<div style="margin-bottom: 0;">{display_emoji} {html_text}</div>'))
            except ImportError:
                # Fallback if IPython is somehow missing
                print(f"{display_emoji} {voice_text}")
        else:
            # Standard terminal output gets the clean text without URL clutter
            print(f"{display_emoji} {voice_text}")

        # Check if the user has globally enabled voice. Default is '1' (On) for fresh installs
        voice_enabled = self.db.get('voice_enabled', '1') == '1'
        
        if not voice_enabled:
            return # Exit early, the print/display statement acts as the visual fallback

        def _execute_speech():
            if delay > 0:
                import time
                time.sleep(delay)
            try:
                # Import here to avoid circular dependencies
                from imports.voice_synthesis import chip_voice_system
                if chip_voice_system and chip_voice_system.voice_ready:
                     # Acoustic Sanitization (Using the voice_text so it doesn't read URLs!)
                     safe_text = voice_text.replace('\u0329', '')
                     safe_text = safe_text.replace('—', ', ').replace('–', ', ')
                     
                     # This blocks the current thread while playing
                     chip_voice_system.speak_text(safe_text)
            except Exception:
                pass

        # If we don't want to wait, or if we have a delay, spawn a background thread
        if not wait or delay > 0:
            import threading
            threading.Thread(target=_execute_speech, daemon=True).start()
        else:
            _execute_speech()

    def mute(self, verbose: bool = True):
        """Instantly silence the AI and disable future voice output globally."""
        self.db['voice_enabled'] = '0'
        try:
            from imports.voice_synthesis import chip_voice_system
            if chip_voice_system:
                chip_voice_system.stop_speaking()
        except Exception:
            pass
        if verbose:
            print("🔇 Global mute activated. The machine will be quiet.")

    def unmute(self, verbose: bool = True):
        """Enable AI voice output globally."""
        self.db['voice_enabled'] = '1'
        if verbose:
            print("🔊 Global voice enabled. The machine will speak.")

    def voice_controls(self):
        """Display an interactive widget in the Jupyter Notebook to toggle voice on/off."""
        if not self.is_notebook_context:
            print("Voice controls are optimized for the Notebook context.")
            return

        try:
            import ipywidgets as widgets
            from IPython.display import display, clear_output
        except ImportError:
            print("⚠️ ipywidgets is required for interactive controls. Use pip.mute() or pip.unmute() instead.")
            return

        # Match the global default of '1' (On)
        current_state = self.db.get('voice_enabled', '1') == '1'

        toggle = widgets.ToggleButton(
            value=current_state,
            description='🔊 Voice ON' if current_state else '🔇 Voice OFF',
            button_style='success' if current_state else 'danger',
            tooltip='Toggle the global voice synthesis system'
        )
        
        out = widgets.Output()

        def on_toggle(change):
            with out:
                clear_output(wait=True)
                new_state = change['new']
                if new_state:
                    self.unmute(verbose=True)
                    toggle.description = '🔊 Voice ON'
                    toggle.button_style = 'success'
                else:
                    self.mute(verbose=True)
                    toggle.description = '🔇 Voice OFF'
                    toggle.button_style = 'danger'

        toggle.observe(on_toggle, 'value')
        display(widgets.VBox([toggle, out]))

    def imperio(self, side_quest=False, emoji: str = "🌀", newline: bool = True):
        """
        The Compulsion. Finalizes a step and ushers the user to the next.
        If a side_quest is active, the machine demands external action 
        before the spell can continue.

        Args:
            side_quest: bool or str. 
                - True: Mandatory action required.
                - "optional": Action available but not required.
                - "release": The workflow is complete; lift the compulsion.
                - False: Standard progression.
            emoji: The icon prepended to the spoken text.
            newline: If True (default), prints an empty line before the compulsion for visual pacing.
        """
        if newline:
            print()

        if side_quest is True:
            self.speak(
                "You must complete the side-quest instructions above before running the next cell.",
                emoji=emoji
            )
        elif side_quest == "optional":
            self.speak(
                "Go on the optional side-quest above. Then run the next cell to continue.",
                emoji=emoji
            )
        elif side_quest == "release":
            self.speak(
                "Workflow complete.",
                emoji=emoji
            )
        else:
            self.speak("Done step. Run the next cell.", emoji=emoji)

    def show_llm_optics(self, target_url: str):
        """Displays a numbered, alphabetical list of files and a button to open the cache directory."""
        from tools.scraper_tools import get_safe_path_component
        import ipywidgets as widgets
        from IPython.display import display
        
        domain, url_path_slug = get_safe_path_component(target_url)
        cache_dir = self.paths.browser_cache / domain / url_path_slug

        if cache_dir.exists():
            print(f"📁 Contents of {cache_dir}:\n")
            print("Let's examine the artifacts I extracted. Click the button to open the folder on your computer...")
            
            # 1. Filter for files and sort them alphabetically by name
            files = sorted([f for f in cache_dir.iterdir() if f.is_file()], key=lambda x: x.name.lower())
            
            # 2. Iterate with an index for numbering
            for i, item in enumerate(files, 1):
                size_kb = item.stat().st_size / 1024
                print(f"{i}. {item.name} ({size_kb:.1f} KB)")
                    
            # Create the "Open Folder" button
            button = widgets.Button(
                description=f"📂 Open Folder",
                tooltip=f"Open {cache_dir.resolve()}",
                button_style='success'
            )
            
            def on_button_click(b):
                self.open_folder(str(cache_dir))
                
            button.on_click(on_button_click)
            display(button)
        else:
            print("Directory not found. The scrape may not have completed successfully.")

    def register_dialogue(self, dialogue_dict: dict):
        """Registers a dictionary of narrative scripts into the self's memory."""
        self.dialogue_tree.update(dialogue_dict)

    def make_singular(self, word):
        """Convert a potentially plural word to its singular form using simple rules.

        This uses basic suffix replacement rules to handle common English plurals.
        It's designed for the 80/20 rule - handling common cases without complexity.

        Args:
            word (str): The potentially plural word to convert

        Returns:
            str: The singular form of the word
        """
        word = word.strip()
        if not word:
            return word
        if word.lower() in ('data', 'media', 'series', 'species', 'news'):
            return word
        irregulars = {'children': 'child', 'people': 'person', 'men': 'man', 'women': 'woman', 'teeth': 'tooth', 'feet': 'foot', 'geese': 'goose', 'mice': 'mouse', 'criteria': 'criterion'}
        if word.lower() in irregulars:
            return irregulars[word.lower()]
        if word.lower().endswith('ies'):
            return word[:-3] + 'y'
        if word.lower().endswith('ves'):
            return word[:-3] + 'f'
        if word.lower().endswith('xes') or word.lower().endswith('sses') or word.lower().endswith('shes') or word.lower().endswith('ches'):
            return word[:-2]
        if word.lower().endswith('s') and (not word.lower().endswith('ss')):
            return word[:-1]
        return word


    def set_chat(self, chat_instance):
            """Set the chat instance after initialization."""
            self.chat = chat_instance


    def get_message_queue(self):
        """Return the message queue instance for ordered message delivery."""
        return self.message_queue

    def step_button(self, visual_step_number: str, preserve: bool = False, revert_label: str = None) -> str:
        """
        Formats the revert button text.
        Uses visual_step_number for "Step X" numbering if revert_label is not provided.

        Args:
            visual_step_number: The visual step number (e.g., "1", "2", "3") based on position in workflow
            preserve: Whether to use the preserve symbol (⟲) instead of revert symbol (↶)
            revert_label: Custom label to use instead of "Step X" format
        """
        symbol = '⟲' if preserve else '↶'
        if revert_label:
            button_text = f'{symbol}\xa0{revert_label}'
        else:
            button_text = f'{symbol}\xa0Step\xa0{visual_step_number}'
        return button_text

    def get_ui_constants(self):
        """Access centralized UI constants through dependency injection."""
        return CFG.UI_CONSTANTS

    def get_config(self):
        """Access centralized configuration through dependency injection."""
        return CFG

    def get_button_border_radius(self):
        """Get the global button border radius setting."""
        return CFG.UI_CONSTANTS['BUTTON_STYLES']['BORDER_RADIUS']

    def register_workflow_routes(self, plugin_instance):
        """
        Register standard and step-specific routes for a workflow plugin.

        This helper extracts the common route registration boilerplate from workflow __init__ methods,
        while maintaining the WET principle - each workflow still explicitly calls this method.

        Args:
            plugin_instance: The workflow plugin instance with app, APP_NAME, and steps attributes
        """
        app_name = plugin_instance.APP_NAME
        steps = plugin_instance.steps

        # Standard workflow lifecycle routes
        routes = [
            (f'/{app_name}/init', plugin_instance.init, ['POST']),
            (f'/{app_name}/revert', plugin_instance.handle_revert, ['POST']),
            (f'/{app_name}/unfinalize', plugin_instance.unfinalize, ['POST'])
        ]

        # Dynamically create routes for each step from the plugin's steps list
        for step_obj in steps:
            step_id = step_obj.id
            handler_method = getattr(plugin_instance, step_id, None)
            if handler_method:
                current_methods = ['GET']
                if step_id == 'finalize':
                    current_methods.append('POST')
                routes.append((f'/{app_name}/{step_id}', handler_method, current_methods))

            # Only data steps (not 'finalize') have explicit _submit handlers
            if step_id != 'finalize':
                submit_handler_method = getattr(plugin_instance, f'{step_id}_submit', None)
                if submit_handler_method:
                    routes.append((f'/{app_name}/{step_id}_submit', submit_handler_method, ['POST']))

        # Register all routes with the FastHTML app
        for path, handler, *methods_list_arg in routes:
            current_methods = methods_list_arg[0] if methods_list_arg else ['GET']
            plugin_instance.app.route(path, methods=current_methods)(handler)

    async def log_api_call_details(self, pipeline_id: str, step_id: str, call_description: str, http_method: str, url: str, headers: dict, payload: Optional[dict] = None, response_status: Optional[int] = None, response_preview: Optional[str] = None, response_data: Optional[dict] = None, curl_command: Optional[str] = None, python_command: Optional[str] = None, estimated_rows: Optional[int] = None, actual_rows: Optional[int] = None, file_path: Optional[str] = None, file_size: Optional[str] = None, notes: Optional[str] = None):
        """Log complete API call details for extreme observability and Jupyter reproduction.

        This provides the same level of transparency for API calls as is used in BQL query logging,
        including copy-paste ready Python code for Jupyter notebook reproduction.
        """
        log_entry_parts = []
        log_entry_parts.append(f'  [API Call] {call_description or "API Request"}')
        log_entry_parts.append(f'  Pipeline ID: {pipeline_id}')
        log_entry_parts.append(f'  Step ID: {step_id}')
        log_entry_parts.append(f'  Method: {http_method}')
        log_entry_parts.append(f'  URL: {url}')
        if headers:
            headers_preview = {k: v for k, v in headers.items() if k.lower() not in ['authorization', 'cookie', 'x-api-key']}
            if len(headers_preview) != len(headers):
                headers_preview['<REDACTED_AUTH_HEADERS>'] = f'{len(headers) - len(headers_preview)} hidden'
            # Use Rich JSON display for headers
            pretty_headers = slog.rich_json_display(headers_preview, title="API Headers", console_output=True, log_output=True)
            log_entry_parts.append(f'  Headers: {pretty_headers}')
        if payload:
            try:
                # Use Rich JSON display for payload
                pretty_payload = slog.rich_json_display(payload, title="API Payload", console_output=True, log_output=True)
                log_entry_parts.append(f'  Payload:\n{pretty_payload}')
            except Exception:
                log_entry_parts.append(f'  Payload: {payload}')
        if curl_command:
            log_entry_parts.append(f'  cURL Command:\n{curl_command}')
        if python_command:
            # Use centralized emoji configuration for console messages
            python_emoji = CFG.UI_CONSTANTS['EMOJIS']['PYTHON_CODE']
            snippet_emoji = CFG.UI_CONSTANTS['EMOJIS']['CODE_SNIPPET']
            comment_divider = CFG.UI_CONSTANTS['CODE_FORMATTING']['COMMENT_DIVIDER']
            snippet_intro = CFG.UI_CONSTANTS['CONSOLE_MESSAGES']['PYTHON_SNIPPET_INTRO'].format(
                python_emoji=python_emoji,
                snippet_emoji=snippet_emoji
            )
            snippet_end = CFG.UI_CONSTANTS['CONSOLE_MESSAGES']['PYTHON_SNIPPET_END'].format(
                python_emoji=python_emoji,
                snippet_emoji=snippet_emoji
            )
            # Add Python snippet with complete BEGIN/END block
            log_entry_parts.append(f'{snippet_intro}\n{python_command}')
            log_entry_parts.append('# Note: The API token should be loaded from a secure file location.')
            log_entry_parts.append(f'{comment_divider}')
            log_entry_parts.append(f'{snippet_end}')
        if estimated_rows is not None:
            log_entry_parts.append(f'  Estimated Rows (from pre-check): {estimated_rows:,}')
        if actual_rows is not None:
            log_entry_parts.append(f'  Actual Rows Downloaded: {actual_rows:,}')
        if response_status is not None:
            log_entry_parts.append(f'  Response Status: {response_status}')
        if response_preview:
            try:
                parsed = json.loads(response_preview)
                # Use Rich JSON display for response preview
                pretty_preview = slog.rich_json_display(parsed, title="API Response Preview", console_output=True, log_output=True)
                log_entry_parts.append(f'  Response Preview:\n{pretty_preview}')
            except Exception:
                log_entry_parts.append(f'  Response Preview:\n{response_preview}')

        # Enhanced transparency for discovery endpoints - log full response data
        is_discovery_endpoint = self._is_discovery_endpoint(url)
        if response_data and is_discovery_endpoint:
            try:
                # Use Rich JSON display for discovery response
                pretty_response = slog.rich_json_display(response_data, title=f"🔍 Discovery Response: {call_description}", console_output=True, log_output=True)
                log_entry_parts.append(f'  🔍 FULL RESPONSE DATA (Discovery Endpoint):\n{pretty_response}')

            except Exception as e:
                log_entry_parts.append(f'  🔍 FULL RESPONSE DATA (Discovery Endpoint): [Error formatting JSON: {e}]\n{response_data}')

                # Still display in console even if JSON formatting fails
                slog.console.print(f"❌ Discovery Response Error: {e}", style="red")
                slog.console.print(f"Raw data: {str(response_data)}", style="dim")

        if file_path:
            log_entry_parts.append(f'  Associated File Path: {file_path}')
        if file_size:
            log_entry_parts.append(f'  Associated File Size: {file_size}')
        if notes:
            log_entry_parts.append(f'  Notes: {notes}')

        full_log_message = '\n'.join(log_entry_parts)
        logger.info(f'\n🚀 === API CALL TRANSPARENCY ===\n{full_log_message}\n🚀 === END API TRANSPARENCY ===')
        is_bql = 'bql' in (call_description or '').lower() or 'botify query language' in (call_description or '').lower()

    def _is_discovery_endpoint(self, url: str) -> bool:
        """Detect if this is a key discovery endpoint that should have full response logging.

        Args:
            url: The API endpoint URL

        Returns:
            bool: True if this is a discovery endpoint that should log full response data
        """
        discovery_patterns = [
            '/analyses/',  # Covers both /analyses/{username}/{project}/light and regular analyses
            '/advanced_export',  # Field discovery endpoint
        ]

        return any(pattern in url for pattern in discovery_patterns)

    async def log_mcp_call_details(self, operation_id: str, tool_name: str, operation_type: str, mcp_block: str = None, request_payload: Optional[dict] = None, response_data: Optional[dict] = None, response_status: Optional[int] = None, external_api_url: Optional[str] = None, external_api_method: str = 'GET', external_api_headers: Optional[dict] = None, external_api_payload: Optional[dict] = None, external_api_response: Optional[dict] = None, external_api_status: Optional[int] = None, execution_time_ms: Optional[float] = None, notes: Optional[str] = None):
        """Log complete MCP operation details for extreme observability and Jupyter reproduction.

        This provides the same level of transparency for MCP operations as the BQL query logging,
        including copy-paste ready Python code for external API reproduction.

        Args:
            operation_id: Unique identifier for this MCP operation
            tool_name: Name of the MCP tool being executed
            operation_type: Type of operation (tool_execution, api_call, etc.)
            mcp_block: Raw MCP block that triggered the operation
            request_payload: Payload sent to MCP tool executor
            response_data: Response from MCP tool executor
            response_status: HTTP status from MCP tool executor
            external_api_url: URL of external API called (if any)
            external_api_method: HTTP method for external API
            external_api_headers: Headers sent to external API
            external_api_payload: Payload sent to external API
            external_api_response: Response from external API
            external_api_status: HTTP status from external API
            execution_time_ms: Total execution time in milliseconds
            notes: Additional context or notes
        """
        log_entry_parts = []
        log_entry_parts.append(f'  [MCP Operation] {operation_type.title()} - {tool_name}')
        log_entry_parts.append(f'  Operation ID: {operation_id}')
        log_entry_parts.append(f'  Tool Name: {tool_name}')
        log_entry_parts.append(f'  Operation Type: {operation_type}')
        log_entry_parts.append(f'  Timestamp: {self.get_timestamp()}')

        if execution_time_ms is not None:
            log_entry_parts.append(f'  Execution Time: {execution_time_ms:.2f}ms')

        # MCP Block Details
        if mcp_block:
            log_entry_parts.append(f'  MCP Block:')
            # Indent each line of the MCP block for better readability
            indented_block = '\n'.join(f'    {line}' for line in mcp_block.strip().split('\n'))
            log_entry_parts.append(indented_block)

        # Internal MCP Tool Executor Request
        if request_payload:
            log_entry_parts.append('')  # Extra space for visual separation
            log_entry_parts.append(f'  MCP Tool Executor Request:')
            log_entry_parts.append(f'    URL: http://127.0.0.1:5001/mcp-tool-executor')
            log_entry_parts.append(f'    Method: POST')
            try:
                # Use Rich JSON display for MCP request payload
                pretty_payload = slog.rich_json_display(request_payload, title="MCP Tool Executor Request", console_output=True, log_output=True)
                # Indent the JSON for consistency
                indented_payload = '\n'.join(f'    {line}' for line in pretty_payload.split('\n'))
                log_entry_parts.append(f'    Payload:\n{indented_payload}')
            except Exception:
                log_entry_parts.append(f'    Payload: {request_payload}')

        # Internal MCP Tool Executor Response
        if response_data or response_status:
            log_entry_parts.append('')  # Extra space for visual separation
            log_entry_parts.append(f'  MCP Tool Executor Response:')
            if response_status:
                log_entry_parts.append(f'    Status: {response_status}')
            if response_data:
                try:
                    # Use Rich JSON display for MCP response data
                    pretty_response = slog.rich_json_display(response_data, title="MCP Tool Executor Response", console_output=True, log_output=True)
                    # Indent the JSON for consistency
                    indented_response = '\n'.join(f'    {line}' for line in pretty_response.split('\n'))
                    log_entry_parts.append(f'    Response:\n{indented_response}')
                except Exception:
                    log_entry_parts.append(f'    Response: {response_data}')

        # External API Call Details (the actual external service)
        if external_api_url:
            log_entry_parts.append('')  # Extra space for visual separation
            log_entry_parts.append(f'  External API Call:')
            log_entry_parts.append(f'    URL: {external_api_url}')
            log_entry_parts.append(f'    Method: {external_api_method}')

            if external_api_headers:
                # Redact sensitive headers
                headers_preview = {k: v for k, v in external_api_headers.items() if k.lower() not in ['authorization', 'cookie', 'x-api-key']}
                if len(headers_preview) != len(external_api_headers):
                    headers_preview['<REDACTED_AUTH_HEADERS>'] = f'{len(external_api_headers) - len(headers_preview)} hidden'
                # Use Rich JSON display for external API headers
                pretty_headers = slog.rich_json_display(headers_preview, title="External API Headers", console_output=True, log_output=True)
                log_entry_parts.append(f'    Headers: {pretty_headers}')

            if external_api_payload:
                try:
                    # Use Rich JSON display for external API payload
                    pretty_payload = slog.rich_json_display(external_api_payload, title="External API Payload", console_output=True, log_output=True)
                    indented_payload = '\n'.join(f'    {line}' for line in pretty_payload.split('\n'))
                    log_entry_parts.append(f'    Payload:\n{indented_payload}')
                except Exception:
                    log_entry_parts.append(f'    Payload: {external_api_payload}')

            if external_api_status:
                log_entry_parts.append(f'    Response Status: {external_api_status}')

            if external_api_response:
                try:
                    # Use Rich JSON display for external API response
                    pretty_response = slog.rich_json_display(external_api_response, title="External API Response", console_output=True, log_output=True)
                    indented_response = '\n'.join(f'    {line}' for line in pretty_response.split('\n'))
                    log_entry_parts.append(f'    Response:\n{indented_response}')
                except Exception:
                    log_entry_parts.append(f'    Response: {external_api_response}')

        # Generate copy-paste ready Python code for Jupyter reproduction
        if external_api_url:
            python_code = self._generate_mcp_python_code(
                tool_name=tool_name,
                external_api_url=external_api_url,
                external_api_method=external_api_method,
                external_api_headers=external_api_headers,
                external_api_payload=external_api_payload,
                operation_id=operation_id
            )

            # Use centralized emoji configuration for console messages
            python_emoji = CFG.UI_CONSTANTS['EMOJIS']['PYTHON_CODE']
            snippet_emoji = CFG.UI_CONSTANTS['EMOJIS']['CODE_SNIPPET']
            comment_divider = CFG.UI_CONSTANTS['CODE_FORMATTING']['COMMENT_DIVIDER']
            snippet_intro = CFG.UI_CONSTANTS['CONSOLE_MESSAGES']['PYTHON_SNIPPET_INTRO'].format(
                python_emoji=python_emoji,
                snippet_emoji=snippet_emoji
            )
            snippet_end = CFG.UI_CONSTANTS['CONSOLE_MESSAGES']['PYTHON_SNIPPET_END'].format(
                python_emoji=python_emoji,
                snippet_emoji=snippet_emoji
            )

            # Add Python snippet with complete BEGIN/END block and visual separation
            log_entry_parts.append('')  # Extra space before Python code
            log_entry_parts.append(f'{snippet_intro}')
            log_entry_parts.append(f'{comment_divider}')
            log_entry_parts.append(f'{python_code}')
            log_entry_parts.append('# Note: This code reproduces the external API call made by the MCP tool.')
            log_entry_parts.append(f'{comment_divider}')
            log_entry_parts.append(f'{snippet_end}')
            log_entry_parts.append('')  # Extra space after Python code

        if notes:
            log_entry_parts.append(f'  Notes: {notes}')

        full_log_message = '\n'.join(log_entry_parts)
        logger.info(f'\n🚀 === MCP OPERATION TRANSPARENCY ===\n{full_log_message}\n🚀 === END MCP TRANSPARENCY ===')

    def _generate_mcp_python_code(self, tool_name: str, external_api_url: str, external_api_method: str = 'GET', external_api_headers: Optional[dict] = None, external_api_payload: Optional[dict] = None, operation_id: str = None) -> str:
        """Generate copy-paste ready Python code for reproducing MCP external API calls in Jupyter.

        This mirrors the pattern used in BQL query logging but for MCP operations.
        """
        lines = []
        lines.append(f'# MCP Tool Reproduction: {tool_name}')
        if operation_id:
            lines.append(f'# Operation ID: {operation_id}')
        lines.append(f'# Generated at: {self.get_timestamp()}')
        lines.append('')
        lines.append('import aiohttp')
        lines.append('import asyncio')
        lines.append('import json')
        lines.append('from pprint import pprint')
        lines.append('')
        lines.append('async def reproduce_mcp_call():')
        lines.append('    """Reproduce the external API call made by the MCP tool."""')
        lines.append('    ')
        lines.append(f'    url = "{external_api_url}"')
        lines.append(f'    method = "{external_api_method.upper()}"')
        lines.append('    ')

        # Headers
        if external_api_headers:
            lines.append('    headers = {')
            for key, value in external_api_headers.items():
                if key.lower() in ['authorization', 'cookie', 'x-api-key']:
                    lines.append(f'        "{key}": "REDACTED_FOR_SECURITY",')
                else:
                    lines.append(f'        "{key}": "{value}",')
            lines.append('    }')
        else:
            lines.append('    headers = {}')
        lines.append('    ')

        # Payload
        if external_api_payload and external_api_method.upper() in ['POST', 'PUT', 'PATCH']:
            lines.append('    payload = {')
            try:
                for key, value in external_api_payload.items():
                    if isinstance(value, str):
                        lines.append(f'        "{key}": "{value}",')
                    else:
                        lines.append(f'        "{key}": {json.dumps(value)},')
            except Exception:
                lines.append(f'        # Payload: {external_api_payload}')
            lines.append('    }')
        else:
            lines.append('    payload = None')
        lines.append('    ')

        # Async session and request
        lines.append('    async with aiohttp.ClientSession() as session:')
        if external_api_method.upper() == 'GET':
            lines.append('        async with session.get(url, headers=headers) as response:')
        elif external_api_method.upper() == 'POST':
            lines.append('        async with session.post(url, headers=headers, json=payload) as response:')
        elif external_api_method.upper() == 'PUT':
            lines.append('        async with session.put(url, headers=headers, json=payload) as response:')
        elif external_api_method.upper() == 'DELETE':
            lines.append('        async with session.delete(url, headers=headers) as response:')
        else:
            lines.append(f'        async with session.request("{external_api_method.upper()}", url, headers=headers, json=payload) as response:')

        lines.append('            print(f"Status: {response.status}")')
        lines.append('            print(f"Headers: {dict(response.headers)}")')
        lines.append('            ')
        lines.append('            if response.content_type == "application/json":')
        lines.append('                data = await response.json()')
        lines.append('                print("JSON Response:")')
        lines.append('                pprint(data)')
        lines.append('                return data')
        lines.append('            else:')
        lines.append('                text = await response.text()')
        lines.append('                print("Text Response:")')
        lines.append('                print(text)')
        lines.append('                return text')
        lines.append('')
        divider = CFG.UI_CONSTANTS['CODE_FORMATTING']['COMMENT_DIVIDER']
        lines.append(divider)
        lines.append('# EXECUTION: Choose your environment')
        lines.append(divider)
        lines.append('')
        lines.append('# For Jupyter Notebooks (recommended):')
        lines.append('result = await reproduce_mcp_call()')
        lines.append('print("\\nFinal result:")')
        lines.append('pprint(result)')
        lines.append('')
        lines.append('# For Python scripts (uncomment if needed):')
        lines.append('# if __name__ == "__main__":')
        lines.append('#     result = asyncio.run(reproduce_mcp_call())')
        lines.append('#     print("\\nFinal result:")')
        lines.append('#     pprint(result)')

        return '\n'.join(lines)

    # ========================================
    # REUSABLE BOTIFY PYTHON CODE GENERATION
    # ========================================

    def generate_botify_code_header(self, display_name: str, step_name: str, username: str, project_name: str,
                                    template_info: dict = None, qualifier_config: dict = None) -> list:
        """Generate standardized header for Botify Python debugging code.

        Delegates to external botify_code_generation module to reduce server.py size.
        """
        return botify_code_generation.generate_botify_code_header(
            display_name=display_name,
            step_name=step_name,
            username=username,
            project_name=project_name,
            template_info=template_info,
            qualifier_config=qualifier_config
        )

    def generate_botify_token_loader(self) -> str:
        """Generate the standard Botify token loading function.

        Delegates to external botify_code_generation module to reduce server.py size.
        """
        return botify_code_generation.generate_botify_token_loader()

    def generate_botify_http_client(self, client_name: str, description: str) -> str:
        """Generate the standard HTTP client function for Botify APIs.

        Delegates to external botify_code_generation module to reduce server.py size.
        """
        return botify_code_generation.generate_botify_http_client(client_name, description)

    def generate_botify_main_executor(self, client_function_name: str, api_description: str) -> str:
        """Generate the main execution function for Botify APIs.

        Delegates to external botify_code_generation module to reduce server.py size.
        """
        return botify_code_generation.generate_botify_main_executor(client_function_name, api_description)

    def create_folder_button(self, folder_path: str, icon: str = "📁", text: str = "Open Folder",
                             title_prefix: str = "Open folder") -> object:
        """Generate a standardized folder opening button.

        Centralizes the folder button pattern used across many plugins.
        """
        import urllib.parse

        from fasthtml.common import A

        quoted_path = urllib.parse.quote(str(folder_path))
        title = f"{title_prefix}: {folder_path}"

        return A(
            f"{icon} {text}",
            hx_get=f"/open-folder?path={quoted_path}",
            hx_swap="none",
            title=title,
            cls="button-link"
        )

    # ========================================
    # ADVANCED BOTIFY CODE GENERATION UTILITIES
    # ========================================

    def generate_botify_bqlv2_python_code(self, query_payload, username, project_name, page_size, jobs_payload, display_name, get_step_name_from_payload_func, get_configured_template_func=None, query_templates=None):
        """
        🚀 REUSABLE UTILITY: Generate complete Python code for BQLv2 queries (crawl, GSC)

        Delegates to external botify_code_generation module to reduce server.py size.
        """
        return botify_code_generation.generate_botify_bqlv2_python_code(
            query_payload=query_payload,
            username=username,
            project_name=project_name,
            page_size=page_size,
            jobs_payload=jobs_payload,
            display_name=display_name,
            get_step_name_from_payload_func=get_step_name_from_payload_func,
            get_configured_template_func=get_configured_template_func,
            query_templates=query_templates
        )

    def generate_botify_bqlv1_python_code(self, query_payload, username, project_name, jobs_payload, display_name, get_step_name_from_payload_func):
        """
        🚀 REUSABLE UTILITY: Generate complete Python code for BQLv1 queries (web logs)

        Delegates to external botify_code_generation module to reduce server.py size.
        """
        return botify_code_generation.generate_botify_bqlv1_python_code(
            query_payload=query_payload,
            username=username,
            project_name=project_name,
            jobs_payload=jobs_payload,
            display_name=display_name,
            get_step_name_from_payload_func=get_step_name_from_payload_func
        )

    def get_botify_analysis_path(self, app_name, username, project_name, analysis_slug, filename=None):
        """
        🚀 REUSABLE UTILITY: Construct standardized Botify analysis file paths

        Delegates to external botify_code_generation module to reduce server.py size.
        """
        return botify_code_generation.get_botify_analysis_path(app_name, username, project_name, analysis_slug, filename)

    def fmt(self, endpoint: str) -> str:
        """Format an endpoint string into a human-readable form."""
        if endpoint in self.friendly_names:
            return self.friendly_names[endpoint]
        return title_name(endpoint)

    def _get_clean_state(self, pkey):
        try:
            record = self.pipeline_table[pkey]
            state = json.loads(record.data)
            state.pop('created', None)
            state.pop('updated', None)
            return state
        except (NotFoundError, json.JSONDecodeError):
            return {}

    def get_timestamp(self) -> str:
        return datetime.now().isoformat()

    def get_plugin_context(self, plugin_instance=None):
        """
        Returns the context information about the current plugin and profile.

        Args:
            plugin_instance: Optional plugin instance to extract name from

        Returns:
            dict: Contains plugin_name, profile_id, and profile_name
        """
        profile_id = self.get_current_profile_id()
        profile_name = self.get_profile_name()
        plugin_name = None
        display_name = None
        if plugin_instance:
            if hasattr(plugin_instance, 'DISPLAY_NAME'):
                display_name = plugin_instance.DISPLAY_NAME
            if hasattr(plugin_instance, 'name'):
                plugin_name = plugin_instance.name
            elif hasattr(plugin_instance, '__class__'):
                plugin_name = plugin_instance.__class__.__name__
            if plugin_name and (not display_name):
                if plugin_name in self.friendly_names:
                    display_name = self.friendly_names[plugin_name]
                else:
                    display_name = title_name(plugin_name)
        return {'plugin_name': display_name or plugin_name, 'internal_name': plugin_name, 'profile_id': profile_id, 'profile_name': profile_name}

    @pipeline_operation
    def initialize_if_missing(self, pkey: str, initial_step_data: dict = None) -> tuple[Optional[dict], Optional[Card]]:
        try:
            state = self.read_state(pkey)
            if state:
                return (state, None)
            now = self.get_timestamp()
            state = {'created': now, 'updated': now}
            if initial_step_data:
                app_name = None
                if 'app_name' in initial_step_data:
                    app_name = initial_step_data.pop('app_name')
                state.update(initial_step_data)
            self.pipeline_table.insert({'pkey': pkey, 'app_name': app_name if app_name else None, 'data': json.dumps(state), 'created': now, 'updated': now})
            return (state, None)
        except:
            error_card = Card(H3('ID Already In Use'), P(f"The ID '{pkey}' is already being used by another workflow. Please try a different ID."), style=self.id_conflict_style())
            return (None, error_card)

    def read_state(self, pkey: str) -> dict:
        logger.debug(f'Reading state for pipeline: {pkey}')
        try:
            record = self.pipeline_table[pkey]

            # 🎯 UNIFIED FIX: Handle both dicts (from notebook) and objects (from server).
            if record:
                if isinstance(record, dict) and 'data' in record:
                    # Handle dictionary from notebook context
                    state = json.loads(record['data'])
                    return state
                elif hasattr(record, 'data'):
                    # Handle object from server context
                    state = json.loads(record.data)
                    return state

            return {}
        except NotFoundError:
            # This is the expected exception when a record doesn't exist.
            logger.debug(f'No record found for pkey: {pkey}')
            return {}
        except Exception as e:
            logger.debug(f'Error reading state for pkey {pkey}: {str(e)}')
            return {}

    def write_state(self, pkey: str, state: dict) -> None:
        state['updated'] = datetime.now().isoformat()
        payload = {'pkey': pkey, 'data': json.dumps(state), 'updated': state['updated']}
        # Use Rich JSON display for debug payload
        formatted_payload = slog.rich_json_display(payload, console_output=False, log_output=True)
        logger.debug(f'Update payload:\n{formatted_payload}')
        self.pipeline_table.update(payload)
        verification = self.read_state(pkey)
        # Use Rich JSON display for verification
        formatted_verification = slog.rich_json_display(verification, console_output=False, log_output=True)
        logger.debug(f'Verification read:\n{formatted_verification}')

    def format_links_in_text(self, text):
        """
        Convert plain URLs in text to clickable HTML links.
        Safe for logging but renders as HTML in the UI.
        """
        url_pattern = 'https?://(?:[-\\w.]|(?:%[\\da-fA-F]{2}))+'

        def replace_url(match):
            url = match.group(0)
            return f'<a href="{url}" target="_blank">{url}</a>'
        return re.sub(url_pattern, replace_url, text)

    async def stream(self, message, **kwargs):
        """Wrapper that delegates to the external stream orchestrator."""
        from imports.stream_orchestrator import stream_orchestrator # Import just-in-time
        # Correctly pass self (pipulate_instance), self.chat (chat_instance), and message
        return await stream_orchestrator(self, self.chat, message, **kwargs)

    async def _handle_llm_stream(self):
        """Handles the logic for an interruptible LLM stream."""
        try:
            await self.chat.broadcast('%%STREAM_START%%')
            conversation_history = self.append_to_conversation()
            response_text = ''
    
            logger.info("ORCHESTRATOR: Entering LLM stream loop.")
            async for chunk in self.process_llm_interaction(self.active_local_model, conversation_history):
                await self.chat.broadcast(chunk)
                response_text += chunk
            logger.info(f"ORCHESTRATOR: Exited LLM stream loop. Full response_text: '{response_text}'")
        except asyncio.CancelledError:
            logger.info("LLM stream was cancelled by user.")
        except Exception as e:
            logger.error(f'Error in LLM stream: {e}', exc_info=True)
            raise
        finally:
            await self.chat.broadcast('%%STREAM_END%%')
            logger.debug("LLM stream finished or cancelled, sent %%STREAM_END%%")

    def display_revert_header(self, step_id: str, app_name: str, steps: list, message: str = None, target_id: str = None, revert_label: str = None, remove_padding: bool = False, show_when_finalized: bool = False):
        """Create a UI control for reverting to a previous workflow step.

        The button label uses the visual sequence number of the step.

        Args:
            step_id: The ID of the step to revert to
            app_name: The workflow app name
            steps: List of Step namedtuples defining the workflow
            message: Optional message to display (defaults to step_id)
            target_id: Optional target for HTMX updates (defaults to app container)
            revert_label: Optional custom label for the revert button
            remove_padding: Whether to remove padding from the article (for advanced layout)
            show_when_finalized: Whether to show content when workflow is finalized (default: False for backward compatibility)

        Returns:
            Card: A FastHTML Card component with revert functionality, or None if finalized and show_when_finalized=False
        """
        pipeline_id = self.db.get('pipeline_id', '')
        finalize_step = steps[-1] if steps and steps[-1].id == 'finalize' else None
        if pipeline_id and finalize_step and not show_when_finalized:
            final_data = self.get_step_data(pipeline_id, finalize_step.id, {})
            if finalize_step.done in final_data:
                return None
        step = next((s for s in steps if s.id == step_id), None)
        if not step:
            logger.error(f"Step with id '{step_id}' not found in steps list for display_revert_header.")
            return Div(f'Error: Step {step_id} not found.')
        data_collection_steps = [s for s in steps if s.id != 'finalize']
        visual_step_number = 'N/A'
        try:
            step_index = data_collection_steps.index(step)
            visual_step_number = str(step_index + 1)
        except ValueError:
            logger.warning(f"Step id '{step_id}' (show: '{step.show}') not found in data_collection_steps. Revert button will show 'Step N/A'.")
        refill = getattr(step, 'refill', False)
        target_id = target_id or f'{app_name}-container'
        form = Form(Input(type='hidden', name='step_id', value=step_id), Button(self.step_button(visual_step_number, refill, revert_label), type='submit', cls='button-revert', aria_label=f'Revert to step {visual_step_number}: {step.show}', title=f'Go back to modify {step.show}'), hx_post=f'/{app_name}/revert', hx_target=f'#{target_id}', hx_swap='outerHTML', role='form', aria_label=f'Revert to {step.show} form')
        if not message:
            return form
        article_style = 'display: flex; align-items: center; justify-content: space-between; background-color: var(--pico-card-background-color);'
        if remove_padding:
            article_style += ' padding: 0;'
        return Card(Div(message, style='flex: 1', role='status', aria_label=f'Step result: {message}'), Div(form, style='flex: 0'), style=article_style, role='region', aria_label=f'Step {visual_step_number} controls')

    def display_revert_widget(self, step_id: str, app_name: str, steps: list, message: str = None, widget=None, target_id: str = None, revert_label: str = None, widget_style=None, finalized_content=None, next_step_id: str = None):
        """Create a standardized container for widgets and visualizations.

        Core pattern for displaying rich content below workflow steps with consistent
        styling and DOM targeting for dynamic updates.

        Features:
        - Consistent padding/spacing with revert controls
        - Unique DOM addressing for targeted updates
        - Support for function-based widgets and AnyWidget components
        - Standard styling with override capability
        - Automatic finalized state handling with chain reaction preservation

        Args:
            step_id: ID of the step this widget belongs to
            app_name: Workflow app name
            steps: List of Step namedtuples defining the workflow
            message: Optional message for revert control
            widget: Widget/visualization to display
            target_id: Optional HTMX update target
            revert_label: Optional custom revert button label
            widget_style: Optional custom widget container style
            finalized_content: Content to show when workflow is finalized (if None, uses message with 🔒)
            next_step_id: Next step ID for chain reaction when finalized

        Returns:
            Div: FastHTML container with revert control and widget content, or locked Card when finalized
        """
        # Check if workflow is finalized
        pipeline_id = self.db.get('pipeline_id', '')
        finalize_step = steps[-1] if steps and steps[-1].id == 'finalize' else None
        is_finalized = False
        if pipeline_id and finalize_step:
            final_data = self.get_step_data(pipeline_id, finalize_step.id, {})
            is_finalized = finalize_step.done in final_data

        if is_finalized:
            # Create locked view for finalized workflow
            step = next((s for s in steps if s.id == step_id), None)
            step_title = step.show if step else step_id

            if finalized_content is None:
                finalized_content = P(f"Step completed: {message or step_title}")

            locked_card = Card(
                H3(f"🔒 {step_title}", role='heading', aria_level='3'),
                Div(finalized_content, cls='custom-card-padding-bg', role='status', aria_label=f'Finalized content for {step_title}'),
                role='region',
                aria_label=f'Finalized step: {step_title}'
            )

            # Add next step trigger if provided
            if next_step_id:
                return Div(
                    locked_card,
                    Div(id=next_step_id, hx_get=f'/{app_name}/{next_step_id}', hx_trigger='load'),
                    id=step_id
                )
            else:
                return Div(locked_card, id=step_id)

        # Normal revert widget behavior for non-finalized workflows
        revert_row = self.display_revert_header(step_id=step_id, app_name=app_name, steps=steps, message=message, target_id=target_id, revert_label=revert_label, remove_padding=True)
        if widget is None or revert_row is None:
            return revert_row

        # Use CSS class for widget content styling, allow custom widget_style override
        widget_container_attrs = {
            'id': f'{step_id}-widget-{hash(str(widget))}',
            'role': 'region',
            'aria_label': f'Widget content for {step_id}'
        }
        if widget_style:
            widget_container_attrs['style'] = widget_style
        else:
            widget_container_attrs['cls'] = 'widget-content'

        return Div(revert_row, Div(widget, **widget_container_attrs), id=f'{step_id}-content', cls='card-container', role='article', aria_label=f'Step content: {step_id}')

    def tree_display(self, content):
        """Create a styled display for file paths in tree or box format.

        Example widget function demonstrating reusable, styled components
        with consistent spacing in workflow displays.

        Args:
            content: Content to display (tree-formatted or plain path)

        Returns:
            Pre: Styled Pre component
        """
        is_tree = '\n' in content and ('└─' in content or '├─' in content)
        if is_tree:
            return Pre(content, cls='tree-display-tree')
        else:
            return Pre(content, cls='tree-display-path')

    def finalized_content(self, message: str, content=None, heading_tag=H4, content_style=None):
        """Create a finalized step display with optional content.

        Companion to display_revert_widget_advanced for finalized workflows,
        providing consistent styling for both states.

        Args:
            message: Message to display (typically with 🔒 lock icon)
            content: FastHTML component to display below message
            heading_tag: Tag to use for message (default: H4)
            content_style: Optional custom content container style

        Returns:
            Card: FastHTML Card component for finalized state
        """
        if content is None:
            return Card(message)

        # Use CSS class for finalized content styling, allow custom content_style override
        content_container_attrs = {}
        if content_style:
            content_container_attrs['style'] = content_style
        else:
            content_container_attrs['cls'] = 'finalized-content'

        return Card(heading_tag(message), Div(content, **content_container_attrs), cls='card-container')

    # START: wrap_with_inline_button
    def wrap_with_inline_button(self, input_element: Input, button_label: str = 'Next ▸', button_class: str = 'primary', show_new_key_button: bool = False, app_name: str = None, **kwargs) -> Div:
        """Wrap an input element with an inline button in a flex container.
    
        Args:
            input_element: The input element to wrap
            button_label: Text to display on the button (default: 'Next ▸')
            button_class: CSS class for the button (default: 'primary')
            show_new_key_button: Whether to show the 🆕 new key button (default: False)
            app_name: App name for new key generation (required if show_new_key_button=True)
            **kwargs: Additional attributes for the button, prefixed with 'button_' (e.g., button_data_testid='my-id')
    
        Returns:
            Div: A flex container with the input and button(s)
        """
        # Styles are now externalized to CSS classes for maintainability
    
        # Generate unique IDs for input-button association
        input_id = input_element.attrs.get('id') or f'input-{hash(str(input_element))}'
        button_id = f'btn-{input_id}'
    
        # Enhance input element with semantic attributes if not already present
        if 'aria_describedby' not in input_element.attrs:
            input_element.attrs['aria_describedby'] = button_id
        if 'id' not in input_element.attrs:
            input_element.attrs['id'] = input_id
    
        # Prepare button attributes with defaults
        button_attrs = {
            'type': 'submit',
            'cls': f'{button_class} inline-button-submit',
            'id': button_id,
            'aria_label': f'Submit {input_element.attrs.get("placeholder", "input")}',
            'title': f'Submit form ({button_label})'
        }
    
        # Process and merge kwargs, allowing overrides
        for key, value in kwargs.items():
            if key.startswith('button_'):
                # Convert button_data_testid to data-testid
                attr_name = key.replace('button_', '', 1).replace('_', '-')
                button_attrs[attr_name] = value
    
        # Create enhanced button with semantic attributes and pass through extra kwargs
        enhanced_button = Button(button_label, **button_attrs)
    
        # Prepare elements for container
        elements = [input_element, enhanced_button]
    
        # Add new key button if requested
        if show_new_key_button and app_name:
            ui_constants = CFG.UI_CONSTANTS
            # 🆕 New Key button styled via CSS class for maintainability
            new_key_button = Button(
                ui_constants['BUTTON_LABELS']['NEW_KEY'],
                type='button',  # Not a submit button
                cls='new-key-button',  # Externalized styling in styles.css
                id=f'new-key-{input_id}',
                hx_get=f'/generate-new-key/{app_name}',
                hx_target=f'#{input_id}',
                hx_swap='outerHTML',
                aria_label='Generate new pipeline key',
                title='Generate a new auto-incremented pipeline key'
            )
            elements.append(new_key_button)
    
        return Div(
            *elements,
            cls='inline-button-container',
            role='group',
            aria_label='Input with submit button' + (' and new key generator' if show_new_key_button else '')
        )
    # END: wrap_with_inline_button

    def create_standard_landing_page(self, plugin_instance):
        """
        Creates a standardized landing page for workflows using centralized UI constants.

        This helper reduces boilerplate while maintaining WET workflow explicitness.
        Each workflow still explicitly calls this method and can customize the result.

        Args:
            plugin_instance: The workflow plugin instance

        Returns:
            Container: Standard landing page structure
        """
        # Standard display name derivation
        try:
            module_file = inspect.getfile(plugin_instance.__class__)
            blank_placeholder_module = importlib.import_module('plugins.910_blank_placeholder')
            derive_public_endpoint_from_filename = blank_placeholder_module.derive_public_endpoint_from_filename
            public_app_name_for_display = derive_public_endpoint_from_filename(Path(module_file).name)
        except (TypeError, AttributeError, ImportError):
            public_app_name_for_display = plugin_instance.APP_NAME

        title = plugin_instance.DISPLAY_NAME or public_app_name_for_display.replace("_", " ").title()

        # Standard pipeline key generation and matching records
        full_key, prefix, _ = self.generate_pipeline_key(plugin_instance)
        self.pipeline_table.xtra(app_name=plugin_instance.APP_NAME)
        matching_records = [record.pkey for record in self.pipeline_table() if record.pkey.startswith(prefix)]

        # Standard form with centralized constants
        ui_constants = CFG.UI_CONSTANTS
        landing_constants = CFG.UI_CONSTANTS['LANDING_PAGE']

        return Container(
            Card(
                H2(title, role='heading', aria_level='2'),
                P(plugin_instance.ENDPOINT_MESSAGE, cls='text-secondary', role='doc-subtitle'),
                Form(
                    self.wrap_with_inline_button(
                        Input(
                            placeholder=landing_constants['INPUT_PLACEHOLDER'],
                            name='pipeline_id',
                            list='pipeline-ids',
                            type='search',
                            required=False,
                            autofocus=True,
                            value=full_key,
                            _onfocus='this.setSelectionRange(this.value.length, this.value.length)',
                            cls='contrast',
                            aria_label='Pipeline ID input',
                            aria_describedby='pipeline-help'
                        ),
                        button_label=ui_constants['BUTTON_LABELS']['ENTER_KEY'],
                        button_class=ui_constants['BUTTON_STYLES']['SECONDARY'],
                        show_new_key_button=True,
                        app_name=plugin_instance.APP_NAME
                    ),
                    self.update_datalist('pipeline-ids', options=matching_records, should_clear=not matching_records),
                    Small('Enter a new ID or select from existing pipelines', id='pipeline-help', cls='text-muted'),
                    hx_post=f'/{plugin_instance.APP_NAME}/init',
                    hx_target=f'#{plugin_instance.APP_NAME}-container',
                    role='form',
                    aria_label=f'Initialize {title} workflow'
                ),
                role='main',
                aria_label=f'{title} workflow landing page'
            ),
            Div(id=f'{plugin_instance.APP_NAME}-container', role='region', aria_label=f'{title} workflow content')
        )

    async def get_state_message(self, pkey: str, steps: list, messages: dict) -> str:
        state = self.read_state(pkey)
        logger.debug(f'\nDEBUG [{pkey}] State Check:')
        # Use Rich JSON display for state debug
        formatted_state = slog.rich_json_display(state, console_output=False, log_output=True)
        logger.debug(formatted_state)
        for step in reversed(steps):
            if step.id not in state:
                continue
            if step.done == 'finalized':
                if step.done in state[step.id]:
                    return self._log_message('finalized', messages['finalize']['complete'])
                return self._log_message('ready to finalize', messages['finalize']['ready'])
            step_data = state[step.id]
            step_value = step_data.get(step.done)
            if step_value:
                msg = messages[step.id]['complete']
                msg = msg.format(step_value) if '{}' in msg else msg
                return self._log_message(f'{step.id} complete ({step_value})', msg)
        return self._log_message('new pipeline', messages['new'])

    def _log_message(self, state_desc: str, message: str) -> str:
        safe_state = state_desc.replace('<', '\\<').replace('>', '\\>')
        safe_message = message.replace('<', '\\<').replace('>', '\\>')
        logger.debug(f'State: {safe_state}, Message: {safe_message}')
        self.append_to_conversation(message, role='system')
        return message

    @pipeline_operation
    def get_step_data(self, pkey: str, step_id: str, default=None) -> dict:
        state = self.read_state(pkey)
        return state.get(step_id, default or {})

    async def clear_steps_from(self, pipeline_id: str, step_id: str, steps: list) -> dict:
        state = self.read_state(pipeline_id)
        start_idx = next((i for i, step in enumerate(steps) if step.id == step_id), -1)
        if start_idx == -1:
            logger.error(f'[clear_steps_from] Step {step_id} not found in steps list')
            return state
        for step in steps[start_idx + 1:]:
            if (not self.PRESERVE_REFILL or not step.refill) and step.id in state:
                logger.debug(f'[clear_steps_from] Removing step {step.id}')
                del state[step.id]
        self.write_state(pipeline_id, state)
        return state

    def id_conflict_style(self):
        """Return style for ID conflict error messages"""
        return 'background-color: #ffdddd; color: #990000; padding: 10px; border-left: 5px solid #990000;'

    def generate_pipeline_key(self, plugin_instance, user_input=None):
        """Generate a standardized pipeline key using the current profile and plugin.

        Creates a composite key in the format: profile_name-plugin_name-user_id
        If user_input is numeric and less than 100, it will be formatted with leading zeros.

        Args:
            plugin_instance: The plugin instance requesting the key
            user_input: Optional user-provided ID part (defaults to auto-incrementing number)

        Returns:
            tuple: (full_key, prefix, user_part) where:
                full_key: The complete pipeline key
                prefix: The profile-plugin prefix
                user_part: The user-specific part of the key
        """
        context = self.get_plugin_context(plugin_instance)
        app_name = getattr(plugin_instance, 'APP_NAME', None)
        plugin_name = app_name or context['plugin_name'] or getattr(plugin_instance, 'DISPLAY_NAME', None) or getattr(plugin_instance, 'app_name', 'unknown')
        profile_name = context['profile_name'] or 'default'
        profile_part = profile_name.replace(' ', '_')
        plugin_part = plugin_name.replace(' ', '_')
        prefix = f'{profile_part}-{plugin_part}-'
        if user_input is None:
            self.pipeline_table.xtra()
            self.pipeline_table.xtra(app_name=app_name)
            app_records = list(self.pipeline_table())
            matching_records = [record.pkey for record in app_records if record.pkey.startswith(prefix)]
            numeric_suffixes = []
            for record_key in matching_records:
                rec_user_part = record_key.replace(prefix, '')
                if rec_user_part.isdigit():
                    numeric_suffixes.append(int(rec_user_part))
            next_number = 1
            if numeric_suffixes:
                next_number = max(numeric_suffixes) + 1
            if next_number < 100:
                user_part = f'{next_number:02d}'
            else:
                user_part = str(next_number)
        elif isinstance(user_input, int) or (isinstance(user_input, str) and user_input.isdigit()):
            number = int(user_input)
            if number < 100:
                user_part = f'{number:02d}'
            else:
                user_part = str(number)
        else:
            user_part = str(user_input)
        full_key = f'{prefix}{user_part}'
        return (full_key, prefix, user_part)

    def parse_pipeline_key(self, pipeline_key):
        """Parse a pipeline key into its component parts.

        Args:
            pipeline_key: The full pipeline key to parse

        Returns:
            dict: Contains profile_part, plugin_part, and user_part components
        """
        parts = pipeline_key.split('-', 2)
        if len(parts) < 3:
            return {'profile_part': parts[0] if len(parts) > 0 else '', 'plugin_part': parts[1] if len(parts) > 1 else '', 'user_part': ''}
        return {'profile_part': parts[0], 'plugin_part': parts[1], 'user_part': parts[2]}

    def update_datalist(self, datalist_id, options=None, should_clear=False):
        """Create a datalist with out-of-band swap for updating dropdown options.

        This helper method allows easy updates to datalist options using HTMX's
        out-of-band swap feature. It can either update with new options or clear all options.

        Args:
            datalist_id: The ID of the datalist to update
            options: List of option values to include, or None to clear
            should_clear: If True, force clear all options regardless of options parameter

        Returns:
            Datalist: A FastHTML Datalist object with out-of-band swap attribute
        """
        if should_clear or options is None:
            return Datalist(id=datalist_id, _hx_swap_oob='true')
        else:
            return Datalist(*[Option(value=opt) for opt in options], id=datalist_id, _hx_swap_oob='true')

    def run_all_cells(self, app_name, steps):
        """
        Rebuild the entire workflow UI from scratch.

        This is used after state changes that require the entire workflow to be regenerated,
        such as reverting to a previous step or jumping to a specific step. It's a core
        helper method commonly used in workflow methods like finalize, unfinalize, and
        handle_revert.

        The method creates a fresh container with all step placeholders, allowing
        the workflow to reload from the current state.

        Args:
            app_name: The name of the workflow app
            steps: List of Step namedtuples defining the workflow

        Returns:
            Div: Container with all steps ready to be displayed
        """
        placeholders = []
        for i, step in enumerate(steps):
            trigger = 'load' if i == 0 else None
            placeholders.append(Div(id=step.id, hx_get=f'/{app_name}/{step.id}', hx_trigger=trigger))
        return Div(*placeholders, id=f'{app_name}-container')

    def validate_step_input(self, value, step_show, custom_validator=None):
        """
        Validate step input with default and optional custom validation.

        This helper ensures consistent validation across all workflow steps:
        1. Basic validation: Ensures the input is not empty
        2. Custom validation: Applies workflow-specific validation logic if provided

        When validation fails, it returns an error component ready for direct
        display in the UI, helping maintain consistent error handling.

        Args:
            value: The user input value to validate
            step_show: Display name of the step (for error messages)
            custom_validator: Optional function(value) -> (is_valid, error_msg)

        Returns:
            tuple: (is_valid, error_message, P_component_or_None)
        """
        is_valid = True
        error_msg = ''
        if not value.strip():
            is_valid = False
            error_msg = f'{step_show} cannot be empty'
        if is_valid and custom_validator:
            custom_valid, custom_error = custom_validator(value)
            if not custom_valid:
                is_valid = False
                error_msg = custom_error
        if not is_valid:
            return (False, error_msg, P(error_msg, cls='text-invalid'))
        return (True, '', None)

    async def set_step_data(self, pipeline_id, step_id, step_value, steps, clear_previous=True):
        """
        Update the state for a step and handle reverting.

        This core helper manages workflow state updates, ensuring consistent state 
        management across all workflows. It handles several important tasks:

        1. Clearing subsequent steps when a step is updated (optional)
        2. Storing the new step value in the correct format
        3. Removing any revert target flags that are no longer needed
        4. Persisting the updated state to storage

        Used by workflow step_xx_submit methods to maintain state after form submissions.

        Args:
            pipeline_id: The pipeline key
            step_id: The current step ID
            step_value: The value to store for this step
            steps: The steps list
            clear_previous: Whether to clear steps after this one

        Returns:
            str: The processed step value (for confirmation messages)
        """
        if clear_previous:
            await self.clear_steps_from(pipeline_id, step_id, steps)
        state = self.read_state(pipeline_id)
        step = next((s for s in steps if s.id == step_id), None)
        if step:
            state[step_id] = {step.done: step_value}
            if '_revert_target' in state:
                del state['_revert_target']
            self.write_state(pipeline_id, state)
        return step_value

    def check_finalize_needed(self, step_index, steps):
        """
        Check if we're on the final step before finalization.

        This helper determines if the workflow is ready for finalization by checking
        if the next step in the sequence is the "finalize" step. Workflows use this
        to decide whether to prompt the user to finalize after completing a step.

        Used in step_xx_submit methods to show appropriate finalization prompts
        after the user completes the last regular step in the workflow.

        Args:
            step_index: Index of current step in steps list
            steps: The steps list

        Returns:
            bool: True if the next step is the finalize step
        """
        next_step_id = steps[step_index + 1].id if step_index < len(steps) - 1 else None
        return next_step_id == 'finalize'

    def chain_reverter(self, step_id, step_index, steps, app_name, processed_val):
        """
        Create the standard navigation controls after a step submission.
        This helper generates a consistent UI pattern for step navigation that includes:
        1. A revert control showing the current step's value
        2. An HTMX-enabled div that EXPLICITLY triggers loading the next step using
           hx_trigger="load" (preferred over relying on HTMX event bubbling)
        Now also triggers a client-side event to scroll the main content panel.
        Args:
            step_id: The current step ID
            step_index: Index of current step in steps list
            steps: The steps list
            app_name: The workflow app name
            processed_val: The processed value to display
        Returns:
            HTMLResponse: A FastHTML Div component with revert control and next step trigger,
                          wrapped in an HTMLResponse to include HX-Trigger header.
        """
        step = steps[step_index]
        next_step_id = steps[step_index + 1].id if step_index < len(steps) - 1 else None
        header_component = self.display_revert_header(step_id=step_id, app_name=app_name, message=f'{step.show}: {processed_val}', steps=steps)
        next_step_trigger_div = Div(id=next_step_id, hx_get=f'/{app_name}/{next_step_id}', hx_trigger='load') if next_step_id else Div()
        content_to_swap = Div(header_component, next_step_trigger_div, id=step_id)
        response = HTMLResponse(to_xml(content_to_swap))
        return response

    async def handle_finalized_step(self, pipeline_id, step_id, steps, app_name, plugin_instance=None):
        """
        Handle the case when a step is submitted in finalized state.

        Args:
            pipeline_id: The pipeline key
            step_id: The current step ID
            steps: The steps list
            app_name: The workflow app name
            plugin_instance: Optional plugin instance (for accessing step_messages)

        Returns:
            Div: The rebuilt workflow UI
        """
        state = self.read_state(pipeline_id)
        state[step_id] = {'finalized': True}
        self.write_state(pipeline_id, state)
        step_messages = {}
        if plugin_instance and hasattr(plugin_instance, 'step_messages'):
            step_messages = plugin_instance.step_messages
        message = await self.get_state_message(pipeline_id, steps, step_messages)
        await self.stream(message, verbatim=True)
        return self.run_all_cells(app_name, steps)

    async def finalize_workflow(self, pipeline_id, state_update=None):
        """
        Finalize a workflow by marking it as complete and updating its state.

        Args:
            pipeline_id: The pipeline key
            state_update: Optional additional state to update (beyond finalized flag)

        Returns:
            dict: The updated state
        """
        state = self.read_state(pipeline_id)
        if 'finalize' not in state:
            state['finalize'] = {}
        state['finalize']['finalized'] = True
        state['updated'] = datetime.now().isoformat()
        if state_update:
            state.update(state_update)
        self.write_state(pipeline_id, state)
        return state

    async def unfinalize_workflow(self, pipeline_id):
        """
        Unfinalize a workflow by removing the finalized flag.

        Args:
            pipeline_id: The pipeline key

        Returns:
            dict: The updated state
        """
        state = self.read_state(pipeline_id)
        if 'finalize' in state:
            del state['finalize']
        state['updated'] = datetime.now().isoformat()
        self.write_state(pipeline_id, state)
        return state

    async def process_llm_interaction(self, model_name: str = None, messages: list = None, base_app=None) -> AsyncGenerator[str, None]:
        from rich.table import Table
        from imports.mcp_orchestrator import parse_mcp_request
        import llm
        
        accumulated_response = []
        full_content_buffer = ""
        word_buffer = ""  # Buffer for word-boundary detection
        mcp_detected = False
        chunk_count = 0

        # Match XML/JSON tool tags AND bracket notation commands
        mcp_pattern = re.compile(r'(<mcp-request>.*?</mcp-request>|<tool\s+[^>]*/>|<tool\s+[^>]*>.*?</tool>|\[[^\]]+\])', re.DOTALL)

        # 🎯 GOLDEN PATH EXECUTION MATRIX - ORCHESTRATOR STATUS:
        # ✅ WORKING: XML syntax <tool><params><url>value</url></params></tool>
        # ✅ WORKING: JSON syntax <tool><params>{"url": "value"}</params></tool>
        # 🟡 INTEGRATING: [cmd arg] bracket notation syntax (parser exists, integrating now)
        # 🔴 NOT YET: python -c "..." inline code execution
        # 🔴 NOT YET: python cli.py call forwarding from message stream
        #
        # 🎓 PROGRESSIVE REVEAL DESIGN FOR LLMs (simplest first):
        # Level 1: [mcp-discover] - Ultra-simple for small models
        # Level 2: .venv/bin/python cli.py mcp-discover - Terminal proficiency
        # Level 3: python -c "from imports.ai_tool_discovery_simple_parser import execute_simple_command..."
        # Level 4: <tool name="ai_self_discovery_assistant"><params>{"discovery_type":"capabilities"}</params></tool>
        # Level 5: <tool name="ai_self_discovery_assistant"><params><discovery_type>capabilities</discovery_type></params></tool>
        # Level 6: Formal MCP Protocol - Full conversation loop with automatic tool execution
        #
        # This orchestrator monitors LLM response streams for MCP tool calls.
        # When found, tools are executed asynchronously and results injected back.

        try:
            target_model = model_name or self.active_local_model or CFG.DEFAULT_PROMPT_MODEL
            # 1. THE UNIVERSAL ADAPTER GRABS THE MODEL
            model = llm.get_model(target_model)
            
            # 2. EXTRACT SYSTEM PROMPT AND CONVERSATION
            # llm expects a system prompt, the current prompt, and optionally previous history.
            system_prompt = None
            history = []
            current_prompt = ""
            
            for msg in messages:
                if msg.get('role') == 'system':
                    system_prompt = msg.get('content')
                elif msg == messages[-1] and msg.get('role') == 'user':
                    current_prompt = msg.get('content')
                else:
                    # We map your dicts to llm's expected conversation format if needed, 
                    # but for raw streaming, we can feed the current prompt with history.
                    history.append(msg)

            # 3. TRIGGER THE STREAM
            # We use stream=True. This works identically for OpenAI, Anthropic, Gemini, and Ollama.
            response = model.prompt(
                current_prompt,
                system=system_prompt,
                stream=True
            )
            
            yield '\n'  # Start with a newline for better formatting in UI

            # 4. ITERATE OVER THE UNIVERSAL CHUNKS
            for chunk in response:
                if not chunk:
                    continue
                    
                chunk_count += 1
                
                # If we've already found and handled a tool call, ignore the rest of the stream.
                if mcp_detected:
                    continue

                full_content_buffer += chunk

                # STAGE 3: Active MCP execution - detect and execute formal MCP requests
                formal_mcp_result = parse_mcp_request(full_content_buffer)
                if formal_mcp_result:
                    tool_name, inner_content = formal_mcp_result
                    mcp_detected = True  # Stop streaming the LLM response
                    
                    logger.info(f"🎯 MCP ACTIVATED: Found formal MCP tool call for '{tool_name}'")
                    
                    # Execute the formal MCP tool call
                    fire_and_forget(
                        execute_formal_mcp_tool_call(messages, tool_name, inner_content)
                    )
                    continue  

                # Use regex to find a complete MCP block
                match = mcp_pattern.search(full_content_buffer)
                if match:
                    mcp_block = match.group(1)
                    mcp_detected = True  
                    
                    logger.info(f"🔧 MCP CLIENT: Complete MCP tool call extracted.")

                    fire_and_forget(
                        execute_and_respond_to_tool_call(messages, mcp_block)
                    )
                    continue

                word_buffer += chunk

                # Hold off on yielding if we might be building a tool call
                if '<tool' in word_buffer or '<mcp-request' in word_buffer or '```xml' in word_buffer:
                    continue

                parts = re.split(r'(\s+)', word_buffer)
                if len(parts) > 1:
                    complete_parts = parts[:-1]
                    word_buffer = parts[-1]
                    for part in complete_parts:
                        accumulated_response.append(part)
                        yield part
                        
                # ⚡ CRITICAL: Because the `llm` stream is technically synchronous under the hood, 
                # we yield to the asyncio event loop to keep the UI perfectly responsive.
                await asyncio.sleep(0)

            # Flush the remaining buffer
            if word_buffer and not mcp_detected:
                accumulated_response.append(word_buffer)
                yield word_buffer

            # Final logging table for LLM responses (including tool calls)
            if accumulated_response:
                final_response = ''.join(accumulated_response)
                table = Table(title='Chat Response')
                table.add_column('Accumulated Response')
                table.add_row(final_response, style='green')
                slog.print_and_log_table(table, "LLM RESPONSE - ")

        except llm.errors.NeedsKeyException:
            error_msg = f'Authentication missing for {target_model}. Please check your .env file or API keys.'
            logger.error(f"🔍 DEBUG: {error_msg}")
            yield error_msg
        except Exception as e:
            error_msg = f'Error: {str(e)}'
            logger.error(f"🔍 DEBUG: Unexpected error in process_llm_interaction: {e}")
            yield error_msg

    def read(self, job: str) -> dict:
        """Reads the entire state dictionary for a given job (pipeline_id)."""
        state = self.read_state(job)
        state.pop('created', None)
        state.pop('updated', None)
        return state
    
    def write(self, job: str, state: dict):
        """Writes an entire state dictionary for a given job (pipeline_id)."""
        existing_state = self.read_state(job)
        if 'created' in existing_state:
            state['created'] = existing_state['created']
        self.write_state(job, state)

    def clear_memory(self, job: str):
        """Completely obliterates a job's state from the database."""
        try:
            self.pipeline_table.delete(job)
            logger.info(f"🗑️ Memory wiped for job: {job}")
        except Exception as e:
            logger.debug(f"Memory wipe failed or job not found for {job}: {e}")
    
    def set(self, job: str, step: str, value: any):
        """Sets a key-value pair within a job's state for notebook usage."""
        state = self.read_state(job)
        if not state:
            state = {'created': self.get_timestamp()}
    
        state[step] = value
        state['updated'] = self.get_timestamp()
    
        payload = {
            'pkey': job,
            'app_name': 'notebook',
            'data': json.dumps(state),
            'created': state.get('created', state['updated']),
            'updated': state['updated']
        }
        self.pipeline_table.upsert(payload, pk='pkey')
    
    def get(self, job: str, step: str, default: any = None) -> any:
        """Gets a value for a key within a job's state."""
        state = self.read_state(job)
        return state.get(step, default)

    async def scrape(self, 
                     url: str, 
                     take_screenshot: bool = False, 
                     mode: str = 'selenium', 
                     headless: bool = True, 
                     verbose: bool = True, 
                     persistent: bool = False, 
                     override_cache: bool = False,
                     profile_name: str = "default", 
                     delay_range: tuple = None, **kwargs):
        """
        Gives AI "eyes" by performing browser automation or HTTP requests to scrape a URL.

        This method is the primary entrypoint for scraping and supports multiple modes.
        The default mode is 'selenium' which uses a full browser.

        Args:
            url (str): The URL to scrape.
            take_screenshot (bool): Whether to capture a screenshot (selenium mode only). Defaults to False.
            mode (str): The scraping mode to use ('selenium', 'requests', etc.). Defaults to 'selenium'.
            headless (bool): Whether to run the browser in headless mode (selenium mode only). Defaults to True.
            persistent (bool): Whether to use a persistent browser profile. Defaults to False.
            profile_name (str): The name of the persistent profile to use. Defaults to "default".
            override_cache (bool): If True, deletes existing cached artifacts before scraping. Defaults to False.
            delay_range (tuple): A tuple (min, max) for random delay in seconds between requests.
            **kwargs: Additional parameters to pass to the underlying automation tool.

        Returns:
            dict: The result from the scraper tool, including paths to captured artifacts.
        """
        from tools.scraper_tools import selenium_automation
        from urllib.parse import urlparse, quote
        from datetime import datetime

        logger.info(f"👁️‍🗨️ Initiating scrape for: {url} (Mode: {mode}, Headless: {headless}, Persistent: {persistent})")

        # --- New Directory Logic ---
        parsed_url = urlparse(url)
        domain = parsed_url.netloc
        path = parsed_url.path or '/'
        # Use quote with an empty safe string to encode everything, including slashes
        url_path_slug = quote(path, safe='')

        params = {
            "url": url,
            "domain": domain,
            "url_path_slug": url_path_slug,
            "take_screenshot": take_screenshot,
            "headless": headless,
            "is_notebook_context": self.is_notebook_context, # Pass the context flag
            "verbose": verbose,
            "persistent": persistent,
            "override_cache": override_cache,
            "profile_name": profile_name,
            "delay_range": delay_range,
            **kwargs # Pass through any other params
        }

        if mode == 'selenium':
            try:
                result = await selenium_automation(params)
                return result
            except Exception as e:
                logger.error(f"❌ Advanced scrape failed for {url}: {e}")
                return {"success": False, "error": str(e)}
        else:
            logger.warning(f"Scrape mode '{mode}' is not yet implemented.")
            return {"success": False, "error": f"Mode '{mode}' not implemented."}

    def _find_project_root(self, start_path):
        """Walks up from a starting path to find the project root (marked by 'flake.nix')."""
        current_path = Path(start_path).resolve()
        while current_path != current_path.parent:
            if (current_path / 'flake.nix').exists():
                return current_path
            current_path = current_path.parent
        return None

    def nbup(self, notebook_filename: str, modules: tuple = None):
        """
        Cleans and syncs a notebook and optionally its associated Python modules
        from the working 'Notebooks/' directory back to the version-controlled
        'assets/nbs/' template directory.
        """
        # Import necessary libraries inside the function
        import nbformat
        from pathlib import Path
        import os
        import shutil
        import ast
        import astunparse # Our new dependency
        import re # Need to import 're' for the regex part

        ### INPUT PROCESSING START ###
        # Ensure the notebook filename has the .ipynb extension
        if not notebook_filename.endswith(".ipynb"):
            notebook_filename = f"{notebook_filename}.ipynb"
        ### INPUT PROCESSING END ###

        ### NEW LOGIC STARTS HERE ###
        class SecretScrubber(ast.NodeTransformer):
            """An AST transformer to replace string literals in assignments with safe defaults or None."""
            def visit_Assign(self, node):
                # Check if the value being assigned is a string constant
                if isinstance(node.value, ast.Constant) and isinstance(node.value.value, str):
                    # Identify the target variable names of this assignment
                    target_names = [t.id for t in node.targets if isinstance(t, ast.Name)]
                    
                    if "CLIENT_DOMAIN" in target_names:
                        node.value = ast.Constant(value="uhnd.com")
                    elif "BOTIFY_PROJECT_URL" in target_names:
                        node.value = ast.Constant(value="https://app.botify.com/uhnd-com/uhnd.com-demo-account/")
                    else:
                        # Replace all other string values with None
                        node.value = ast.Constant(value=None)
                return node
        ### NEW LOGIC ENDS HERE ###

        # --- Define Sample Data for Scrubbing ---
        SAMPLE_FILTERS_SOURCE = [
            "# --- Define Custom Excel Tab Filters --- \n",
            "# (This list is scrubbed by pip.nbup() and returned to this default)\n",
            "\n",
            "targeted_filters = [\n",
            "    (\"Gifts\", ['gift', 'gifts', 'idea', 'ideas', 'present', 'presents', 'give', 'giving', 'black friday', 'cyber monday', 'cyber week', 'bfcm', 'bf', 'cm', 'holiday', 'deals', 'sales', 'offer', 'discount', 'shopping']),\n",
            "    (\"Broad Questions\", '''am are can could did do does for from had has have how i is may might must shall should was were what when where which who whom whose why will with would'''.split()),\n",
            "    (\"Narrow Questions\", '''who whom whose what which where when why how'''.split()),\n",
            "    (\"Popular Modifiers\", ['how to', 'best', 'review', 'reviews']),\n",
            "    (\"Near Me\", ['near me', 'for sale', 'nearby', 'closest', 'near you', 'local'])\n",
            "]\n",
            "\n",
            "pip.set(job, 'targeted_filters', targeted_filters)\n",
            "print(f\"✅ Stored {len(targeted_filters)} custom filter sets in pip state.\")"
        ]
        SAMPLE_PROMPT_SOURCE_FAQ = [
            "**Your Role (AI Content Strategist):**\n",
            "\n",
            "You are an AI Content Strategist. \n",
            "Make 5 Frequently Asked Questions for each page.\n",
            "For each question, produce the following so it fits the data structure:\n",
            "\n",
            "1. priority: integer (1-5, 1 is highest)\n",
            "2. question: string (The generated question)\n",
            "3. target_intent: string (What is the user's goal in asking this?)\n",
            "4. justification: string (Why is this a valuable question to answer? e.g., sales, seasonal, etc.)"
        ]

        SAMPLE_PROMPT_SOURCE_URLI = [
            "**Your Role (SEO URL Auditor):**\n",
            "\n",
            "Based on the input data for a single webpage (URL, title, h1s, h2s, status code, and markdown body), provide the following:\n",
            "\n",
            "1.  **ai_selected_keyword**: The single best keyword phrase (2-5 words) the page appears to be targeting. Prioritize the \`title\` and \`h1_tags\` for this selection.\n",
            "2.  **ai_score**: On a scale of 1-5 (5 is best), how well-aligned the page's content (\`title\`, \`h1s\`, \`h2s\`, \`markdown\`) is to this single keyword. A 5 means the keyword is used effectively and naturally in key places. A 1 means it's barely present.\n",
            "3.  **keyword_rationale**: A brief (1-sentence) rationale for the score, including the user's most likely search intent (Informational, Commercial, Navigational, or Transactional)."
        ]

        PROMPT_MAP = {
            "FAQuilizer": SAMPLE_PROMPT_SOURCE_FAQ,
            "URLinspector": SAMPLE_PROMPT_SOURCE_URLI,
        }

        SAMPLE_URL_LIST_SOURCE = [
            "# Enter one URL per line\n",
            "https://nixos.org/     # Linux\n",
            "https://jupyter.org/   # Python\n",
            "https://neovim.io/     # vim\n",
            "https://git-scm.com/   # git\n",
            "https://www.fastht.ml/ # FastHTML\n"
            'https://pipulate.com/  # AIE (Pronounced "Ayyy")'
        ]

        # 1. Find the project root in a portable way
        project_root = self._find_project_root(os.getcwd())
        if not project_root:
            print("❌ Error: Could not find project root (flake.nix). Cannot sync.")
            return

        # --- Notebook Sync Logic ---
        print(f"🔄 Syncing notebook '{notebook_filename}'...")
        notebook_source_path = project_root / "Notebooks" / notebook_filename
        notebook_dest_path = project_root / "assets" / "nbs" / notebook_filename

        if not notebook_source_path.exists():
            print(f"❌ Error: Source notebook not found at '{notebook_source_path}'")
        else:
            try:
                with open(notebook_source_path, 'r', encoding='utf-8') as f:
                    nb = nbformat.read(f, as_version=4)

                # --- Scrub proprietary data ---
                notebook_base_name = Path(notebook_filename).stem
                prompt_source_to_use = PROMPT_MAP.get(notebook_base_name, SAMPLE_PROMPT_SOURCE_FAQ)

                for cell in nb.cells:
                    tags = cell.metadata.get("tags", [])
                    if "prompt-input" in tags:
                        cell.source = prompt_source_to_use
                        print(f"    ✓ Scrubbed and replaced 'prompt-input' cell using prompt for '{notebook_base_name}'.")
                    elif "url-list-input" in tags:
                        cell.source = SAMPLE_URL_LIST_SOURCE
                        print("    ✓ Scrubbed and replaced 'url-list-input' cell.")
                    elif "custom-filters-input" in tags:
                        cell.source = SAMPLE_FILTERS_SOURCE
                        print("    ✓ Scrubbed and replaced 'custom-filters-input' cell.")

                    ### NEW LOGIC STARTS HERE ###
                    elif "secrets" in tags and cell.cell_type == 'code':
                        try:
                            # First, do the robust AST-based scrubbing for variable assignments
                            tree = ast.parse(cell.source)
                            scrubber = SecretScrubber()
                            transformed_tree = scrubber.visit(tree)
                            scrubbed_source = astunparse.unparse(transformed_tree)
                            # Then, run the existing regex for pip.api_key for backward compatibility
                            cell.source = re.sub(
                                r'(key\s*=\s*)["\'].*?["\']',
                                r'\1None',
                                scrubbed_source
                            )
                            print("    ✓ Scrubbed variable assignments in 'secrets' cell.")
                        except SyntaxError:
                            print("    ⚠️ Could not parse 'secrets' cell, falling back to regex only.")
                            # Fallback to just the regex if AST fails
                            cell.source = re.sub(r'(key\s*=\s*)["\'].*?["\']', r'\1None', cell.source)
                    ### NEW LOGIC ENDS HERE ###

                # --- Existing Cleaning Logic ---
                original_cell_count = len(nb.cells)
                pruned_cells = [
                    cell for cell in nb.cells if 'pip.nbup' not in cell.source
                ]
                if len(pruned_cells) < original_cell_count:
                    print("    ✓ Auto-pruned the 'pip.nbup()' command cell from the template.")
                nb.cells = pruned_cells

                for cell in nb.cells:
                    if cell.cell_type == 'code':
                        # Normalize cell source to string if it's a list
                        source_text = cell.source
                        if isinstance(source_text, list):
                            source_text = "".join(source_text)
                        
                        # This regex is still needed for calls not in a 'secrets' cell
                        if "secrets" not in cell.metadata.get("tags", []):
                            cell.source = re.sub(r'(key\s*=\s*)["\'].*?["\']', r'\1None', source_text)

                        # Clear outputs and execution counts
                        cell.outputs.clear()
                        cell.execution_count = None
                        if 'metadata' in cell and 'execution' in cell.metadata:
                            del cell.metadata['execution']

                with open(notebook_dest_path, 'w', encoding='utf-8') as f:
                    nbformat.write(nb, f)
                print(f"✅ Success! Notebook '{notebook_filename}' has been cleaned and synced.")

            except Exception as e:
                print(f"❌ An error occurred during the notebook sync process: {e}")

        # --- Module Sync Logic (remains unchanged) ---
        if modules:
            print("\n--- Syncing Associated Modules ---")
            if isinstance(modules, str): modules = (modules,)
            for module_name in modules:
                module_filename = f"{module_name}.py"
                module_source_path = project_root / "Notebooks" / "imports" / module_filename # Look inside imports/
                module_dest_path = project_root / "assets" / "nbs" / "imports" / module_filename # Sync back to imports/
                if module_source_path.exists():
                    try:
                        shutil.copy2(module_source_path, module_dest_path)
                        print(f"    🧬 Synced module: '{module_filename}'")
                    except Exception as e:
                        print(f"    ❌ Error syncing module '{module_filename}': {e}")
                else:
                    print(f"    ⚠️ Warning: Module file not found, skipping sync: '{module_source_path}'")


    def prompt(self, prompt_text: str, model_name: str = None, system_prompt: str = None):
        """
        The Universal Adapter prompt. 
        Sends a single, one-shot prompt to ANY configured AI model (local or cloud)
        with zero vendor lock-in.
        """
        import llm
        model_name = model_name or CFG.DEFAULT_PROMPT_MODEL
        
        print(f"🤖 Channeling intent through {model_name}...")

        try:
            # The Magic Wand invokes the Universal Adapter
            model = llm.get_model(model_name)

            # --- BULLETPROOF AUTHENTICATION ---
            # Bypass internal keystore mapping issues by explicitly assigning the key
            env_var_name = None
            if 'claude' in model_name.lower() or 'anthropic' in model_name.lower():
                env_var_name = 'ANTHROPIC_API_KEY'
            elif 'gpt' in model_name.lower() or 'openai' in model_name.lower():
                env_var_name = 'OPENAI_API_KEY'
            elif 'groq' in model_name.lower():
                env_var_name = 'GROQ_API_KEY'
            elif 'gemini' in model_name.lower() or 'google' in model_name.lower():
                env_var_name = 'GEMINI_API_KEY'

            if env_var_name:
                api_key = self.load_secrets(env_var_name)
                if api_key:
                    model.key = api_key
            response = model.prompt(prompt_text, system=system_prompt)
            
            print("✅ Prompt successfully submitted to model. Wait a moment...")
            return response.text()
            
        except llm.errors.NeedsKeyException:
            error_msg = f"❌ Authentication missing for {model_name}. Please set the appropriate key in your .env file."
            print(error_msg)
            return error_msg
        except Exception as e:
            error_msg = f"❌ AI prompt failed: {e}"
            print(error_msg)
            return error_msg

    def resilient_prompt(self, prompt_text: str, system_prompt: str = None, validator: callable = None) -> str:
        """
        The Sovereign Actuator. Executes a prompt across the MODEL_CASCADE with 
        exponential backoff, auto-pulling, and optional AST/logic validation.
        """
        import llm
        import time
        import subprocess
        import os
        
        for model_id in CFG.MODEL_CASCADE:
            attempt = 0
            max_retries = 3
            current_prompt = prompt_text
            
            print(f"🤖 Engaging {model_id}...")
            while attempt < max_retries:
                try:
                    # Auto-pull local models if missing
                    try:
                        model = llm.get_model(model_id)
                    except llm.UnknownModelError:
                        if 'gemini' not in model_id and 'gpt' not in model_id and 'claude' not in model_id:
                            print(f"⏳ Auto-pulling {model_id} via host OS. Please wait...")
                            subprocess.run(['ollama', 'pull', model_id], capture_output=True, check=True)
                            model = llm.get_model(model_id)
                        else:
                            raise

                    # Cloud Auth Gatekeeper
                    if 'gemini' in model_id.lower() and not os.getenv('GEMINI_API_KEY'):
                        self.ensure_credentials('GEMINI_API_KEY', 'Google Gemini')
                        model.key = os.getenv('GEMINI_API_KEY')

                    response = model.prompt(current_prompt, system=system_prompt).text()
                    
                    if validator:
                        is_valid, error_msg = validator(response)
                        if not is_valid:
                            print(f"⚠️ Validation failed on {model_id} (Attempt {attempt+1}): {error_msg.splitlines()[-1]}")
                            current_prompt = f"{prompt_text}\n\nYOUR PREVIOUS RESPONSE FAILED VALIDATION WITH THIS ERROR:\n{error_msg}\n\nPlease correct your formatting and try again."
                            attempt += 1
                            continue
                            
                    return response

                except Exception as e:
                    error_str = str(e).lower()
                    if "quota" in error_str or "resourceexhausted" in error_str:
                        print(f"🛑 Hard Stop: Quota exceeded for {model_id}. Cascading to next model.")
                        break # Break while loop, continue for loop
                    
                    wait_time = 2 ** attempt
                    print(f"⚠️ Network/Execution friction on {model_id}: {e}. Retrying in {wait_time}s...")
                    time.sleep(wait_time)
                    attempt += 1
                    
        print("❌ All models in the cascade exhausted.")
        return None

    def load_secrets(self, key_name="GOOGLE_API_KEY"):
        """
        Orchestrates secret retrieval with a 'Waterfall of Truth':
        1. OS Environment / already loaded .env
        2. Local .env file at project root
        3. Interactive Notebook prompt (UI fallback)
        """
        # Find project root (re-using your existing logic)
        project_root = self._find_project_root(os.getcwd()) or Path.cwd()
        env_path = project_root / ".env"

        # 1 & 2. Try to get from environment or load from .env
        if not os.environ.get(key_name):
            load_dotenv(dotenv_path=env_path)
        
        current_val = os.environ.get(key_name)
        
        if current_val:
            return current_val

        # 3. UI Fallback if we are in a notebook
        if self.is_notebook_context:
            return self._prompt_for_secret(key_name, env_path)
        
        return None

    def _prompt_for_secret(self, key_name, env_path):
        """Interactive widget to capture and persist a secret."""
        import ipywidgets as widgets
        from IPython.display import display, clear_output

        print(f"🔑 {key_name} not found. Let's set it up for this machine.")
        
        password_input = widgets.Password(
            description='API Key:',
            placeholder='Paste key here...',
            style={'description_width': 'initial'}
        )
        button = widgets.Button(description="💾 Save to .env", button_style='success')
        output = widgets.Output()

        def on_click(b):
            val = password_input.value.strip()
            if val:
                # 4. Persistence: Save to .env and set in current session
                with output:
                    clear_output()
                    try:
                        # Ensure .env exists
                        if not env_path.exists():
                            env_path.touch()
                        
                        set_key(str(env_path), key_name, val)
                        os.environ[key_name] = val
                        print(f"✅ Success! {key_name} saved to {env_path.name}")
                        print("You can now continue running your notebook.")
                        # Hide the input after success
                        password_input.close()
                        button.close()
                    except Exception as e:
                        print(f"❌ Error saving key: {e}")

        button.on_click(on_click)
        display(widgets.VBox([password_input, button, output]))
        return None # User will need to run the next cell after clicking save

    # Update your existing api_key method or wrap it
    def api_key(self, job=None, key=None):
        """Standardized wrapper for the new load_secrets logic."""
        # If user hardcoded a key in the cell, we honor it
        if key:
            os.environ["GOOGLE_API_KEY"] = key
            return True
            
        found_key = self.load_secrets("GOOGLE_API_KEY")
        return bool(found_key)

    def reset_credentials(self, env_var_name: str, service_name: str = None):
        """
        Removes a credential from the active environment and the .env vault, 
        then immediately prompts the user to enter a new one via the UI widget.
        
        Args:
            env_var_name (str): The environment variable to remove (e.g., 'BOTIFY_API_TOKEN').
            service_name (str): Friendly name for the UI (e.g., 'Botify'). Auto-derived if None.
        """
        import os
        from dotenv import load_dotenv, set_key
        from IPython.display import clear_output
        
        if not service_name:
            service_name = env_var_name.split('_')[0].title()

        # 1. Clear from active environment
        if env_var_name in os.environ:
            del os.environ[env_var_name]
            
        # 2. Clear from .env vault by setting it to an empty string
        project_root = self._find_project_root(os.getcwd()) or Path.cwd()
        env_path = project_root / ".env"
        
        if env_path.exists():
            set_key(str(env_path), env_var_name, "")
            self.speak(f"{service_name} credentials cleared from the vault.")
        
        # 3. If in a notebook, clear the cell output and pop the widget back up
        if self.is_notebook_context:
            clear_output(wait=True)
            self.ensure_credentials(env_var_name, service_name)
        else:
            print(f"✅ {env_var_name} has been reset. Please update your .env file.")

    def collect_config(self, job: str, config_keys: list):
        """
        Displays a UI widget form to collect or update job-specific configuration.
        Pre-fills inputs with existing values from wand.db if available.
        """
        import ipywidgets as widgets
        from IPython.display import display, clear_output

        if not self.is_notebook_context:
            print(f"ℹ️ collect_config called for job '{job}'. Pre-fill available in Notebooks.")
            return

        # 1. Build the form inputs with pre-filled data
        inputs = {}
        for key in config_keys:
            existing_val = self.get(job, key) or ""
            label = key.replace('_', ' ').title()
            inputs[key] = widgets.Text(
                value=existing_val,
                placeholder=f"Enter {label}...",
                description=f"{label}:",
                style={'description_width': 'initial'},
                layout=widgets.Layout(width='80%')
            )

        submit_btn = widgets.Button(
            description="Update Session Config", 
            button_style='primary',
            icon='save',
            layout=widgets.Layout(width='max-content')
        )
        
        out = widgets.Output()

        def on_submit(b):
            with out:
                clear_output()
                all_valid = True
                updates = []
                for k, widget in inputs.items():
                    val = widget.value.strip()
                    if val:
                        self.set(job, k, val)
                        # Capture both the key and the new value for the read-back
                        updates.append(f"{k} with {val}")
                    else:
                        all_valid = False
                        print(f"❌ '{k}' cannot be empty.")
                
                if all_valid:
                    # Adjust the sentence structure to read naturally
                    self.speak(f"Job '{job}' updated {', '.join(updates)}.")
                    # Keep widgets open so they can see the success, 
                    # but maybe change button color to show it's "saved"
                    submit_btn.button_style = 'success'
                    submit_btn.description = "Config Locked & Saved"
                    self.speak("You may now run the next cell.", emoji="✅")

        submit_btn.on_click(on_submit)
        
        # Display the "Control Panel"
        display(widgets.VBox(list(inputs.values()) + [submit_btn, out]))
        
        # If data already existed, give a silent nudge that they can proceed
        if all(self.get(job, k) for k in config_keys):
            print(f"✅ Existing config detected for '{job}'. You may update above or proceed.")

    def audit_environment(self):
        """
        Reveals the local Python reality and securely masks the .env vault.
        """
        import sys
        import os
        from pathlib import Path
        from dotenv import load_dotenv, dotenv_values
        
        # 1. Force load of .env before auditing to avoid "missing key" false alarms
        project_root = self._find_project_root(Path(__file__).resolve()) or Path.cwd()
        env_path = project_root / '.env'
        if env_path.exists():
            load_dotenv(dotenv_path=env_path)
        
        print("=== 🌍 YOUR LOCAL REALITY ===")
        print(f"🐍 Python Executable: {sys.executable}")
        print(f"📁 Working Directory: {os.getcwd()}")
        print(f"📁 Logs Path: {self.paths.logs}")

        print("\n=== 🛡️ THE VAULT (.env) ===")
        if env_path.exists():
            secrets = dotenv_values(env_path)
            if not secrets:
                print("  (The vault is currently empty)")
            else:
                for key, val in secrets.items():
                    if not val:
                        print(f"  🔑 {key}: [EMPTY]")
                    elif len(val) > 10:
                        # Show first 4 and last 4 chars, mask the rest
                        masked_val = f"{val[:4]}{'*' * (len(val) - 8)}{val[-4:]}"
                        print(f"  🔑 {key}: {masked_val}")
                    else:
                        # Too short to safely show any characters
                        print(f"  🔑 {key}: {'*' * len(val)}")
        else:
            print("  (No .env file found. The vault is unbuilt.)")
        print()

    def import_this(self):
        """
        The Zen of Python, channeled through the Wand.
        Decodes the ancient wisdom and synthesizes it into spoken word.
        """
        import io
        import contextlib
        
        # 1. The 'this' module prints to stdout upon first import.
        # We trap that output in a null buffer to prevent the first duplication.
        with contextlib.redirect_stdout(io.StringIO()):
            import this
            
        # 2. Decode the rot13 cipher natively
        zen_of_python = "".join([this.d.get(c, c) for c in this.s])
        
        # 3. Clean up the text for the TTS engine 
        # (Replace dashes with commas so the voice pauses instead of saying "dash")
        spoken_zen = zen_of_python.replace("--", ",").replace("-", " ")

        # 4. Speak it into reality! 
        # (self.speak automatically handles the console printing, so we don't need manual prints)
        self.speak(
            text=f"Channeling The Zen of Python...\n\n{spoken_zen}",
            delay=0.0,    # Remove the delay so it starts immediately
            wait=True,    # Set wait to True to block the execution of the next cell
            emoji="🐍"
        )

    def verify_local_ai(self, preferred_models: str = None, simulate_state: str = None) -> str:
        """
        Dedicated check for local AI capabilities (Ollama).
        Returns the selected model string if successful, or None if not found,
        while printing clear educational instructions for the user to install it.
        """
        import llm
        import subprocess
        preferred_models = preferred_models or CFG.PREFERRED_LOCAL_MODELS
        
        print("Scanning your system for a local AI brain...")
        try:
            # 1. State Simulation (For Testing)
            if simulate_state == 'no_ollama':
                available_models = []
                has_local = False
            elif simulate_state == 'no_models':
                available_models = ["llama2:7b", "mistral:latest"] # Dummy models, not preferred
                has_local = True
            else:
                # Actual System Check
                available_models = [m.model_id for m in llm.get_models()]
                has_local = any('ollama' in str(type(m)).lower() for m in llm.get_models())
            
            prefs = [p.strip().lower() for p in preferred_models.split(',')]
            selected_local = None
            primary_recommendation = prefs[0].split(':')[0] if prefs else CFG.DEFAULT_PROMPT_MODEL.split(':')[0]
            
            # 2. Fuzzy match for preferred models
            if not simulate_state == 'no_ollama' and not simulate_state == 'no_models':
                for pref in prefs:
                    match = next((m for m in available_models if pref in m.lower() and 'ollama' in str(type(llm.get_model(m))).lower()), None)
                    if match:
                        selected_local = match
                        break
            
            # 3. Response Matrix & Auto-Pull
            if selected_local:
                self.speak(f"Excellent. Local model '{selected_local}' is active and ready.")
                self.db['active_local_model'] = selected_local
                print(f"\n✅ Locked in Local Model: {selected_local}")
                return selected_local
                
            else:
                # Check if the ollama CLI is actually installed on the host OS
                has_cli = False
                try:
                    cli_check = subprocess.run(['ollama', '--version'], capture_output=True, text=True)
                    has_cli = cli_check.returncode == 0
                except FileNotFoundError:
                    pass

                if has_cli:
                    self.speak(f"Ollama is installed, but {primary_recommendation} is missing. Downloading it now. This may take a few minutes.")
                    print(f"\n⏳ Auto-pulling {primary_recommendation} via host OS. Please wait...")
                    
                    # Blocking subprocess call so the notebook waits for the download to finish
                    # capture_output=True silences the massive progress bar in the Jupyter output
                    pull_result = subprocess.run(['ollama', 'pull', primary_recommendation], capture_output=True)

                    if pull_result.returncode == 0:
                        self.speak("Download complete. Model is locked in.")
                        print(f"\n✅ Successfully pulled and locked in Local Model: {primary_recommendation}")
                        self.db['active_local_model'] = primary_recommendation
                        return primary_recommendation
                    else:
                        print(f"\n❌ Failed to pull {primary_recommendation}. Please run 'ollama pull {primary_recommendation}' manually in your terminal.")
                        return None
                else:
                    self.speak("I do not detect a local AI brain. Let's get you set up.")
                    print("\nℹ️ Ollama is not running or not installed.")
                    print("To achieve true Local-First Sovereignty, you need a local AI running on your metal.")
                    print("\n1. Go to https://ollama.com/")
                    print("2. Download and install it for your operating system.")
                    print(f"3. Open your terminal and run: ollama run {primary_recommendation}")
                    print("4. Come back here and re-run this cell.")
                    return None
                
        except Exception as e:
            print(f"❌ Error communicating with the Universal Adapter: {e}")
            return None

    def ensure_credentials(self, env_var_name: str, service_name: str = None, force_prompt: bool = False) -> str:
        """
        The Universal Gatekeeper. Checks for required API keys upfront to prevent 
        mid-workflow lazy-loading crashes. Renders a secure widget if missing.
        
        Args:
            env_var_name: The environment variable to look for (e.g., 'BOTIFY_API_TOKEN').
            service_name: Friendly name for the UI (e.g., 'Botify'). Auto-derived if None.
            force_prompt: If True, ignores cached credentials and forces the UI widget.
        """
        import os
        import ipywidgets as widgets
        from IPython.display import display, clear_output
        from dotenv import load_dotenv, set_key
        
        if not service_name:
            # Auto-derive friendly name (e.g., 'BOTIFY_API_TOKEN' -> 'Botify')
            service_name = env_var_name.split('_')[0].title()

        # 1. Load existing environment variables (override=True fights caching stubbornness)
        project_root = self._find_project_root(os.getcwd()) or Path.cwd()
        env_path = project_root / ".env"
        load_dotenv(dotenv_path=env_path, override=True)
        
        current_val = os.environ.get(env_var_name)
        
        # If force_prompt is True, we deliberately ignore the current value
        if current_val and not force_prompt:
            self.speak(f"{service_name} credentials verified in your environment.")
            print(f"✅ Secure {service_name} connection ready.")
            return current_val

        # 2. Interactive Fallback for Notebooks
        if self.is_notebook_context:
            self.speak(f"Please provide your {service_name} API key.")
            
            key_input = widgets.Password(
                value='',
                placeholder=f'Paste your {service_name} API Key here...',
                description=f'🔑 {service_name}:',
                style={'description_width': 'initial'},
                disabled=False,
                layout=widgets.Layout(width='80%')
            )
            
            submit_btn = widgets.Button(
                description="Save to Vault", 
                button_style='success',
                icon='lock'
            )
            out = widgets.Output()
            
            def on_submit(b):
                with out:
                    clear_output()
                    val = key_input.value.strip()
                    if val:
                        # Save permanently to .env
                        env_path.touch(exist_ok=True)
                        set_key(str(env_path), env_var_name, val)
                        
                        # Export to current runtime environment
                        os.environ[env_var_name] = val
                        
                        # Explicitly set it in Simon Willison's 'llm' tool keychain
                        try:
                            import llm
                            key_alias = env_var_name.split('_')[0].lower()
                            llm.set_key(key_alias, val)
                        except Exception as e:
                            pass # Fail silently if the specific llm set_key implementation differs
                            
                        self.speak(f"{service_name} key securely saved to the vault.")
                        print(f"✅ {env_var_name} successfully encrypted in .env.")
                        
                        # Hide the widget and fire the compulsion to move forward
                        key_input.close()
                        submit_btn.close()
                        self.imperio()
                    else:
                        print("❌ Please enter a valid API key.")
            
            submit_btn.on_click(on_submit)
            display(widgets.VBox([key_input, submit_btn, out]))
            return None
        else:
            print(f"❌ Missing {env_var_name} in environment. Please add it to your .env file.")
            return None


    def verify_cloud_ai(self, preferred_models: str = None, simulate_state: str = None) -> tuple:
        """
        Dedicated check for Cloud AI capabilities.
        Negotiates the preferred cloud model and triggers the credential widget if needed.
        
        Args:
            preferred_models: Comma-separated list of acceptable cloud models.
            simulate_state: For testing. Can be 'missing_key' or None.
            
        Returns:
            tuple: (selected_cloud_model_string, key_ready_boolean)
        """
        import llm
        preferred_models = preferred_models or CFG.PREFERRED_CLOUD_MODELS
        
        print("Scanning Universal Adapter for preferred Cloud models...")
        available_models = [m.model_id for m in llm.get_models()]
        prefs = [p.strip().lower() for p in preferred_models.split(',')]
        selected_cloud = None
        
        for pref in prefs:
            # Find first match that is NOT an ollama model
            match = next((m for m in available_models if pref in m.lower() and 'ollama' not in str(type(llm.get_model(m))).lower()), None)
            if match:
                selected_cloud = match
                break
        
        # Fallback to Gemini if nothing explicit matches, as it's our recommended free tier
        if not selected_cloud:
            selected_cloud = "gemini-1.5-flash-latest" 
            
        print(f"☁️ Selected Cloud Model: {selected_cloud}")
        self.db['active_cloud_model'] = selected_cloud
        
        # Map the selected model to its corresponding environment variable
        env_var_name = None
        if 'claude' in selected_cloud.lower() or 'anthropic' in selected_cloud.lower():
            env_var_name = 'ANTHROPIC_API_KEY'
            service_name = 'Anthropic'
        elif 'gpt' in selected_cloud.lower() or 'openai' in selected_cloud.lower():
            env_var_name = 'OPENAI_API_KEY'
            service_name = 'OpenAI'
        elif 'groq' in selected_cloud.lower():
            env_var_name = 'GROQ_API_KEY'
            service_name = 'Groq'
        else: # Default to Gemini
            env_var_name = 'GEMINI_API_KEY'
            service_name = 'Google Gemini'
            
        # The Gatekeeper: Forces the API key request up-front via a secure widget if missing.
        force_prompt = (simulate_state == 'missing_key')
        key_present = self.ensure_credentials(env_var_name, service_name, force_prompt=force_prompt)
            
        return selected_cloud, bool(key_present)

    def execute_onboarding_airlock(self):
        """
        The Sentinel-Triggered Airlock.
        Crosses the domain boundary to extract data from the Jupyter environment
        and inject it into the FastHTML global state.
        """
        import sqlite3
        import json

        # 1. Explicitly define the path to the Jupyter chrysalis (ignoring current context)
        jupyter_db_path = self.paths.root / "Notebooks" / "data" / "pipeline.sqlite"
        sentinel_path = self.paths.root / "Notebooks" / "data" / ".onboarded"
        
        # 2. Safety Check: Only proceed if the chrysalis exists
        if not jupyter_db_path.exists():
            logger.warning(f"Airlock Aborted: No Jupyter database found at {jupyter_db_path}")
            return False
            
        try:
            logger.info("🌀 AIRLOCK ENGAGED: Extracting Jupyter Onboarding Data...")
            
            # 3. The Surgical Read-Only Extraction
            # Connect in read-only mode to prevent lock conflicts with any hanging Jupyter kernels
            uri = f"file:{jupyter_db_path}?mode=ro"
            conn = sqlite3.connect(uri, uri=True)
            cursor = conn.cursor()
            
            # The data we want is locked inside the 'onboarding_01' pipeline record
            cursor.execute("SELECT data FROM pipeline WHERE pkey = 'onboarding_01'")
            result = cursor.fetchone()
            cursor.execute("SELECT key, value FROM store WHERE key IN ('active_local_model', 'active_cloud_model')")
            global_store_results = cursor.fetchall()
            
            conn.close()
            
            if not result:
                logger.warning("Airlock Empty: No 'onboarding_01' record found.")
                return False
                
            onboarding_state = json.loads(result[0])
            
            # 4. The Transformation & Injection
            
            # Extract Operator Name from pipeline state
            if 'operator_name' in onboarding_state:
                self.db['operator_name'] = onboarding_state['operator_name']
                logger.info(f"🧬 Injected Operator Name: {onboarding_state['operator_name']}")
                
                # Secure it in the subconscious (AI Keychain)
                try:
                    from imports.ai_dictdb import keychain_instance
                    keychain_instance['operator_name'] = onboarding_state['operator_name']
                except Exception as e:
                    logger.warning(f"Could not inject to keychain: {e}")

            # Extract AI Preferences from global store results
            for key, value in global_store_results:
                self.db[key] = value
                if key == 'active_local_model':
                    self.active_local_model = value
                elif key == 'active_cloud_model':
                    self.active_cloud_model = value
                logger.info(f"🧬 Injected {key} from Jupyter store: {value}")

            return True

        except Exception as e:
            logger.error(f"❌ Airlock Failure: {e}")
            return False

    def copy_button(self, text_to_copy: str, label: str = "Copy Markdown"):
        """
        Injects a web-native Copy to Clipboard button directly into the Jupyter output.
        Reuses the global SVG icon for aesthetic consistency.
        """
        if not self.is_notebook_context:
            return # Only render in Jupyter

        import json
        import html
        from IPython.display import display, HTML

        # Grab the exact same SVG used in the Chat UI
        clipboard_svg = CFG.SVG_ICONS.get('CLIPBOARD', '📋')
        
        # Safely escape the text for JS, and then escape the JS for HTML attribute injection
        safe_js_string = json.dumps(text_to_copy)
        safe_html_attr = html.escape(safe_js_string)
        
        # The HTML payload includes inline JS to trigger the browser's native clipboard API
        # Notice we swapped to double-quotes for the onclick attribute to wrap the escaped entities
        html_payload = f"""
        <div style="display: flex; justify-content: flex-end; margin-top: 8px;">
            <button onclick="navigator.clipboard.writeText({safe_html_attr}).then(() => {{ const original = this.innerHTML; this.innerHTML = '✅ Copied!'; setTimeout(() => this.innerHTML = original, 2000); }}).catch(() => this.innerHTML='❌ Error')"
                    style="display: flex; align-items: center; gap: 6px; background: var(--jp-layout-color2, #f5f5f5); border: 1px solid var(--jp-border-color1, #ccc); border-radius: 4px; padding: 6px 12px; cursor: pointer; color: var(--jp-ui-font-color1, #333); font-family: inherit; font-size: 0.9em; transition: all 0.2s;">
                {clipboard_svg} {label}
            </button>
        </div>
        """
        display(HTML(html_payload))
