combined_code.txt

##### my_digital_being/framework/activity_decorator.py #####
import functools
import logging
from typing import Callable, Any, Dict, List, Optional
from datetime import datetime
import json

logger = logging.getLogger(__name__)

def activity(
    name: str,
    energy_cost: float = 0.2,
    cooldown: int = 0,
    required_skills: Optional[List[str]] = None
):
    """Decorator for activity classes."""
    def decorator(cls):
        cls.activity_name = name
        cls.energy_cost = energy_cost
        cls.cooldown = cooldown
        cls.required_skills = required_skills or []
        cls.last_execution = None

        # Add metadata to the class
        cls.metadata = {
            'name': name,
            'energy_cost': energy_cost,
            'cooldown': cooldown,
            'required_skills': required_skills
        }

        # Wrap the execute method
        original_execute = cls.execute

        @functools.wraps(original_execute)
        async def wrapped_execute(self, *args, **kwargs):
            try:
                # Pre-execution checks
                if not self._can_execute():
                    logger.warning(f"Activity {name} is on cooldown")
                    return ActivityResult(
                        success=False,
                        error="Activity is on cooldown"
                    )

                # Log activity start
                logger.info(f"Starting activity: {name}")
                start_time = datetime.now()

                # Execute the activity
                result = await original_execute(self, *args, **kwargs)

                # Post-execution processing
                end_time = datetime.now()
                duration = (end_time - start_time).total_seconds()
                cls.last_execution = end_time

                # Log activity completion
                logger.info(f"Completed activity: {name} in {duration:.2f} seconds")

                return result

            except Exception as e:
                logger.error(f"Error in activity {name}: {e}")
                return ActivityResult(
                    success=False,
                    error=str(e)
                )

        cls.execute = wrapped_execute
        return cls

    return decorator

class ActivityResult:
    """Class to store activity execution results."""
    def __init__(
        self,
        success: bool,
        data: Optional[Any] = None,
        error: Optional[str] = None,
        metadata: Optional[Dict[str, Any]] = None
    ):
        self.success = success
        self.data = data
        self.error = error
        self.metadata = metadata or {}
        self.timestamp = datetime.now()

    def to_dict(self) -> Dict[str, Any]:
        """Convert result to dictionary format."""
        data_dict = {}
        if self.data:
            if hasattr(self.data, 'to_dict'):
                data_dict = self.data.to_dict()
            elif isinstance(self.data, dict):
                data_dict = self.data
            else:
                try:
                    data_dict = json.loads(json.dumps(self.data))
                except:
                    data_dict = str(self.data)

        return {
            'success': self.success,
            'data': data_dict,
            'error': self.error,
            'metadata': self.metadata,
            'timestamp': self.timestamp.isoformat()
        }

    @classmethod
    def success_result(cls, data: Optional[Any] = None, metadata: Optional[Dict[str, Any]] = None):
        """Create a successful result."""
        return cls(success=True, data=data, metadata=metadata)

    @classmethod
    def error_result(cls, error: str, metadata: Optional[Dict[str, Any]] = None):
        """Create an error result."""
        return cls(success=False, error=error, metadata=metadata)

class ActivityBase:
    """Base class for all activities."""
    def __init__(self):
        self.result = None
        self.last_execution: Optional[datetime] = None
        self.cooldown: int = 0

    def _can_execute(self) -> bool:
        """Check if the activity can be executed."""
        if self.last_execution is None:
            return True

        now = datetime.now()
        time_since_last = (now - self.last_execution).total_seconds()
        return time_since_last >= self.cooldown

    def get_result(self) -> Dict[str, Any]:
        """Get the result of the activity execution."""
        if isinstance(self.result, ActivityResult):
            return self.result.to_dict()
        return {
            'success': bool(self.result),
            'data': self.result if self.result else None,
            'error': None,
            'timestamp': datetime.now().isoformat()
        }

    async def execute(self, shared_data) -> ActivityResult:
        """Base execute method that should be overridden by activities."""
        raise NotImplementedError("Activities must implement execute method")

def skill_required(skill_name: str):
    """Decorator to specify required skills for methods."""
    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        def wrapper(self, *args, **kwargs):
            if not hasattr(self, 'required_skills'):
                self.required_skills = []
            if skill_name not in self.required_skills:
                self.required_skills.append(skill_name)
            return func(self, *args, **kwargs)
        return wrapper
    return decorator


##### my_digital_being/framework/activity_loader.py #####
import importlib
import logging
import re
from pathlib import Path
from typing import Dict, Type, Any, Optional

logger = logging.getLogger(__name__)

def read_activity_code(activity_name: str) -> Optional[str]:
    """
    Reads the .py file from 'activities/' by the given filename (e.g. 'activity_tweet.py').
    Returns its text content or None if file not found.
    """
    activity_file = Path(__file__).parent.parent / 'activities' / activity_name
    if not activity_file.exists():
        logger.warning(f"read_activity_code: File not found: {activity_file}")
        return None
    return activity_file.read_text()

def write_activity_code(activity_name: str, new_code: str) -> bool:
    """
    Writes 'new_code' into the .py file in 'activities/' with the given filename.
    Returns True on success, False on error.
    """
    activity_file = Path(__file__).parent.parent / 'activities' / activity_name
    try:
        activity_file.write_text(new_code, encoding='utf-8')
        return True
    except Exception as e:
        logger.error(f"write_activity_code: Failed to write {activity_file}: {e}")
        return False

class ActivityLoader:
    def __init__(self, activities_path: str = None, config: dict = None):
        """
        :param activities_path: Where activity_*.py files live.
        :param config: The main config object from being.configs (used to skip disabled).
        """
        if activities_path is None:
            activities_path = Path(__file__).parent.parent / 'activities'
        self.activities_path = Path(activities_path)

        # [ADDED] We'll read 'activities_config' from config
        self.activities_config = {}
        if config:
            self.activities_config = (
                config.get("activity_constraints", {}).get("activities_config", {})
            )

        self.loaded_activities: Dict[str, Type[Any]] = {}
        logger.info(f"ActivityLoader initialized with path: {self.activities_path}")

    def load_activities(self):
        """Load all activities from the activities directory."""
        if not self.activities_path.exists():
            logger.error(f"Activities directory not found: {self.activities_path}")
            return

        logger.info(f"Starting to load activities from: {self.activities_path}")
        for activity_file in self.activities_path.glob("activity_*.py"):
            try:
                logger.info(f"Found activity file: {activity_file}")
                file_text = activity_file.read_text()

                # We expect a pattern like: class SomeActivity(ActivityBase):
                class_match = re.search(r'class\s+(\w+)\(.*ActivityBase.*\):', file_text)
                if not class_match:
                    logger.error(f"No recognized activity class in {activity_file}")
                    continue

                class_name = class_match.group(1)
                module_name = activity_file.stem  # e.g. "activity_draw"

                # Possibly skip if "enabled": false in activities_config
                activity_cfg = None
                if class_name in self.activities_config:
                    activity_cfg = self.activities_config[class_name]
                elif module_name in self.activities_config:
                    activity_cfg = self.activities_config[module_name]

                if activity_cfg and (activity_cfg.get("enabled") is False):
                    logger.info(f"Activity {class_name} is disabled by config, skipping load.")
                    continue

                spec = importlib.util.spec_from_file_location(module_name, activity_file)
                if spec and spec.loader:
                    module = importlib.util.module_from_spec(spec)
                    spec.loader.exec_module(module)

                    activity_class = getattr(module, class_name)
                    self.loaded_activities[module_name] = activity_class
                    logger.info(f"Successfully loaded activity {module_name} -> class {class_name}")

            except Exception as e:
                logger.error(f"Failed to load activity {activity_file}: {str(e)}", exc_info=True)

    def get_activity(self, activity_name: str) -> Optional[Type[Any]]:
        """Get an activity class by module name (e.g. 'activity_tweet')."""
        return self.loaded_activities.get(activity_name)

    def get_all_activities(self) -> Dict[str, Type[Any]]:
        """Get all loaded activities (module_name -> class)."""
        return self.loaded_activities.copy()

    def reload_activities(self):
        """Reload all activities by clearing and reloading."""
        self.loaded_activities.clear()
        self.load_activities()


##### my_digital_being/framework/activity_selector.py #####
import logging
import random
from typing import Dict, Any, Optional, List, Tuple
from datetime import datetime, timedelta

logger = logging.getLogger(__name__)

class ActivitySelector:
    def __init__(self, constraints: Dict[str, Any], state):
        """
        :param constraints: A dictionary that typically includes:
            {
              "activity_cooldowns": { ... },  # No longer used
              "activity_requirements": { ... },
              "activities_config": { "DrawActivity": {"enabled": false}, ... }
            }
        :param state: The DigitalBeing's State object, used to check mood, energy, etc.
        """
        self.constraints = constraints
        self.state = state

        # Tracks the last time each activity class was executed
        self.last_activity_times: Dict[str, datetime] = {}

        # The loader is not set until set_activity_loader() is called
        self.activity_loader = None

    def set_activity_loader(self, loader):
        """
        Attach an ActivityLoader instance to this ActivitySelector.
        That loader has the loaded_activities dictionary (module_name -> activity_class).
        """
        self.activity_loader = loader
        logger.info("Activity loader set in selector")

    def select_next_activity(self):
        """
        Main entry point:
        1. Gather all available activities (not on cooldown, not disabled).
        2. Filter them by additional requirements like energy, skill requirements, etc.
        3. Use personality to pick one at random (weighted).
        4. Record the time we picked it.
        5. Return the activity instance or None.
        """
        if not self.activity_loader:
            logger.error("Activity loader not set; cannot select activity.")
            return None

        # Step 1: get all activities that are not on cooldown and are enabled
        available_activities = self._get_available_activities()
        if not available_activities:
            next_available = self.get_next_available_times()
            logger.info(f"No activities available at this time. Next available activities: {next_available}")
            return None

        # Step 2: filter out ones that fail "energy" or "activity_requirements"
        suitable_activities = []
        for activity in available_activities:
            activity_name = activity.__class__.__name__
            if (self._check_energy_requirements(activity) and
                    self._check_activity_requirements(activity_name)):
                logger.info(f"Activity {activity_name} is suitable for execution.")
                suitable_activities.append(activity)
            else:
                logger.info(f"Activity {activity_name} does not meet requirements.")

        if not suitable_activities:
            logger.info("No activities suitable for current state.")
            return None

        # Step 3: personality-based selection
        # (If you have a "personality" dict in state, else use {}.)
        personality = self.state.get_current_state().get('personality', {})
        selected_activity = self._select_based_on_personality(
            suitable_activities,
            personality
        )

        if selected_activity:
            chosen_name = selected_activity.__class__.__name__
            logger.info(f"Selected activity: {chosen_name}")
            # Step 4: record the time we picked it
            self.last_activity_times[chosen_name] = datetime.now()

        return selected_activity

    def get_next_available_times(self) -> List[Dict[str, Any]]:
        """
        Provide info on when each loaded activity class will be available again.
        This is mostly for debugging/logging: "You can next run DrawActivity in 1.5 hours", etc.
        We now use the activity's decorator-based cooldown.
        """
        current_time = datetime.now()
        next_available = []

        all_activities = self.activity_loader.get_all_activities()

        for activity_name, activity_class in all_activities.items():
            base_name = activity_class.__name__

            # Pull cooldown from the class (decorator)
            cooldown = getattr(activity_class, 'cooldown', 0)
            last_time = self.last_activity_times.get(base_name)

            if last_time:
                time_since_last = (current_time - last_time).total_seconds()
                time_remaining = max(0, cooldown - time_since_last)
                next_time = current_time + timedelta(seconds=time_remaining)

                next_available.append({
                    'activity': base_name,
                    'available_in_seconds': time_remaining,
                    'next_available_at': next_time.strftime('%Y-%m-%d %H:%M:%S'),
                    'cooldown_period': cooldown
                })
            else:
                # never run before => it's available now
                next_available.append({
                    'activity': base_name,
                    'available_in_seconds': 0,
                    'next_available_at': 'Now',
                    'cooldown_period': cooldown
                })

        # sort by soonest availability
        return sorted(next_available, key=lambda x: x['available_in_seconds'])

    def _get_available_activities(self) -> List[Any]:
        """
        Return a list of *activity instances* that:
          1) Are loaded by the ActivityLoader
          2) Are "enabled" in the config
          3) Are not on cooldown (based on the activity's own decorator-based cooldown)
        Then the caller can further filter them for energy or skill requirements.
        """
        available = []
        current_time = datetime.now()

        all_activities = self.activity_loader.get_all_activities()
        activities_config = self.constraints.get('activities_config', {})

        for module_name, activity_class in all_activities.items():
            base_name = activity_class.__name__

            # 1) skip if disabled
            if base_name in activities_config:
                if activities_config[base_name].get("enabled", True) is False:
                    logger.info(f"Skipping disabled activity: {base_name}")
                    continue

            # 2) check if it's on cooldown
            cooldown = getattr(activity_class, 'cooldown', 0)
            last_time = self.last_activity_times.get(base_name)
            if last_time:
                time_since_last = (current_time - last_time).total_seconds()
                if time_since_last < cooldown:
                    logger.info(f"{base_name} still on cooldown for {cooldown - time_since_last:.1f}s more.")
                    continue

            # If we get here, the activity is enabled & not on cooldown
            try:
                instance = activity_class()
                logger.info(f"Created instance of {base_name} successfully.")
                available.append(instance)
            except Exception as e:
                logger.error(f"Failed to create instance of {base_name}: {e}", exc_info=True)

        return available

    def _check_activity_requirements(self, activity_name: str) -> bool:
        """
        Check constraints['activity_requirements'][activity_name] if you need logic
        for required skills or memory usage. Currently returns True to accept all.
        """
        requirements = self.constraints.get('activity_requirements', {}).get(activity_name, {})
        logger.debug(f"Checking requirements for {activity_name}: {requirements}")
        return True

    def _check_energy_requirements(self, activity) -> bool:
        """
        Check if the being has enough energy for the activity (activity.energy_cost).
        """
        current_energy = self.state.get_current_state().get('energy', 1.0)
        required_energy = getattr(activity, 'energy_cost', 0.2)
        has_energy = current_energy >= required_energy

        if not has_energy:
            logger.info(f"Insufficient energy for {activity.__class__.__name__} "
                        f"(required={required_energy}, current={current_energy}).")
        return has_energy

    def _select_based_on_personality(self, activities: List[Any], personality: Dict[str, float]) -> Optional[Any]:
        """
        Given a list of candidate activity instances, choose one with a weighted random approach.
        """
        if not activities:
            return None

        weights = []
        for activity in activities:
            weight = 1.0
            if hasattr(activity, 'creativity_factor'):
                weight *= (1 + personality.get('creativity', 0.5) * activity.creativity_factor)
            if hasattr(activity, 'social_factor'):
                weight *= (1 + personality.get('friendliness', 0.5) * activity.social_factor)
            weights.append(weight)

        chosen = random.choices(activities, weights=weights, k=1)[0]
        return chosen


##### my_digital_being/framework/api_key_setup.py #####
"""Tool for securely setting up API keys."""
import logging
from typing import List, Dict, Tuple
import os
from .secret_storage import secret_manager

logger = logging.getLogger(__name__)

class APIKeySetup:
    """Manages the setup and validation of API keys for skills."""

    @staticmethod
    async def setup_keys(skill_name: str, required_keys: List[str]) -> Dict[str, bool]:
        """
        Set up API keys for a skill using the configured secret storage.

        Args:
            skill_name: Name of the skill requiring API keys
            required_keys: List of required API key names

        Returns:
            Dictionary mapping key names to setup success status
        """
        results = {}

        try:
            # For Replit environment, use ask_secrets
            if 'REPL_ID' in os.environ:
                from replit import ask_secrets
                env_keys = [f"{skill_name.upper()}_{key.upper()}_API_KEY" for key in required_keys]

                await ask_secrets(
                    secret_keys=env_keys,
                    user_message=f"""
The {skill_name} skill requires the following API keys to function:
{', '.join(required_keys)}

Please provide these keys to enable the skill's functionality.
These will be stored securely as environment variables.
"""
                )

            # Verify keys were set properly
            for key in required_keys:
                exists = await secret_manager.check_api_key_exists(skill_name, key)
                results[key] = exists

                if exists:
                    logger.info(f"Successfully set up {key} API key for {skill_name}")
                else:
                    logger.warning(f"Failed to set up {key} API key for {skill_name}")

        except Exception as e:
            logger.error(f"Error setting up API keys for {skill_name}: {e}")
            for key in required_keys:
                results[key] = False

        return results

    @staticmethod
    async def check_skill_keys(skill_name: str, required_keys: List[str]) -> Tuple[bool, List[str]]:
        """
        Check if a skill has all required API keys configured.

        Args:
            skill_name: Name of the skill to test
            required_keys: List of required API keys

        Returns:
            Tuple of (success, list of missing keys)
        """
        missing_keys = []
        for key in required_keys:
            exists = await secret_manager.check_api_key_exists(skill_name, key)
            if not exists:
                missing_keys.append(key)

        return len(missing_keys) == 0, missing_keys

    @staticmethod
    async def list_skill_requirements(skill_requirements: Dict[str, List[str]]) -> str:
        """
        Get a formatted string of all skills and their API key requirements.

        Args:
            skill_requirements: Dictionary mapping skill names to their required keys

        Returns:
            Formatted string showing all skills and their required API keys
        """
        if not skill_requirements:
            return "No skills with API key requirements registered."

        output = ["Skill API Key Requirements:"]
        for skill, keys in skill_requirements.items():
            success, missing = await APIKeySetup.check_skill_keys(skill, keys)
            status = "✓" if success else "✗"
            output.append(f"\n{status} {skill}:")
            for key in keys:
                exists = await secret_manager.check_api_key_exists(skill, key)
                configured = "✓" if exists else "✗"
                output.append(f"  {configured} {key}")

        return "\n".join(output)

##### my_digital_being/framework/api_management.py #####
"""
Unified API key management system with flexible storage backends.
Implements:
 - get_skill_status -> returns any "required_keys"
 - get_composio_integrations -> calls composio_manager.list_available_integrations()
 - set_api_key -> if you want to store them
"""

import logging
from typing import Dict, Any, Optional, Set, List

from .secret_storage import secret_manager
from .composio_integration import composio_manager

logger = logging.getLogger(__name__)

class APIManager:
    def __init__(self):
        # Example: track required keys for each skill
        self._required_keys: Dict[str, Set[str]] = {}
        self._secret_manager = secret_manager
        self._composio_manager = composio_manager
        logger.info("Initialized API Manager with Composio integration")

    @property
    def composio_manager(self):
        return self._composio_manager

    def register_required_keys(self, skill_name: str, required_keys: List[str]) -> bool:
        """
        Register that a given skill_name requires the specified list of key names (e.g. ["OPENAI"]).
        """
        if not skill_name or not required_keys:
            return False
        self._required_keys[skill_name] = set(required_keys)
        logger.info(f"Registered keys for skill {skill_name}: {required_keys}")
        return True

    def get_required_keys(self, skill_name: Optional[str] = None) -> Dict[str, List[str]]:
        """
        Return a dict of skill -> list of required keys.
        If skill_name is provided, return only that one skill's key list.
        """
        if skill_name:
            # Return just one skill's keys if it exists
            if skill_name in self._required_keys:
                return { skill_name: list(self._required_keys[skill_name]) }
            else:
                return { skill_name: [] }
        else:
            # Return all
            return { skill: list(keys) for skill, keys in self._required_keys.items() }

    async def check_api_key_exists(self, skill_name: str, key_name: str) -> bool:
        """
        Pass-through to secret_manager to check if a key is set.
        This is used in e.g. ImageGenerationSkill or other activities that do:
        `await api_manager.check_api_key_exists(...).`
        """
        return await self._secret_manager.check_api_key_exists(skill_name, key_name)

    async def get_api_key(self, skill_name: str, key_name: str) -> Optional[str]:
        """
        Return the actual API key string from secret_manager.
        Called by skill_chat.py's initialize() or skill_generate_image.py, etc.
        """
        return await self._secret_manager.get_api_key(skill_name, key_name)

    async def get_skill_status(self) -> Dict[str, Any]:
        """
        Example: For each skill, show which keys are configured or not.
        """
        skills_status = {}
        for skill, keys in self._required_keys.items():
            skill_info = {
                "display_name": skill.title(),
                "required_keys": {}
            }
            for k in keys:
                # Check if configured
                exists = await self._secret_manager.check_api_key_exists(skill, k)
                skill_info["required_keys"][k] = bool(exists)
            skills_status[skill] = skill_info
        return skills_status

    async def set_api_key(self, skill_name: str, key_name: str, value: str) -> Dict[str, Any]:
        """
        Store a new API key into secret_manager for a given skill & key name.
        """
        success = await self._secret_manager.set_api_key(skill_name, key_name, value)
        return {"success": success, "affected_skills": {}}

    async def get_composio_integrations(self) -> List[Dict[str, Any]]:
        """
        Ask the ComposioManager for the available integrations (connected or not).
        """
        return await self._composio_manager.list_available_integrations()

    async def list_actions_for_app(self, app_name: str) -> Dict[str, Any]:
        """
        Calls composio_manager.list_actions_for_app(app_name).
        """
        return await self._composio_manager.list_actions_for_app(app_name)

# Global
api_manager = APIManager()


##### my_digital_being/framework/composio_integration.py #####
"""
Composio integration module for managing OAuth flows and dynamic tool integration.
Implements:
 - handle_oauth_callback(...) to finalize the connection
 - store a connected indicator in _oauth_connections
 - list_available_integrations() returns "connected": True if we have that.
 - list_actions_for_app(...) returns the app's actions by calling Composio's API directly

[ADDED] We now persist these connections in ./storage/composio_oauth.json
"""

import os
import logging
import json
from pathlib import Path
from typing import Dict, Any, List

import requests  # Used for the direct Composio API call

from .secret_storage import secret_manager
from composio_openai import ComposioToolSet

logger = logging.getLogger(__name__)

class ComposioManager:
    def __init__(self):
        self._toolset = None
        self._entity_id = "MyDigitalBeing"
        self._oauth_connections: Dict[str, Dict[str, Any]] = {}
        self._available_apps: Dict[str, Any] = {}

        # [ADDED] We store the OAuth connections in a JSON file
        self.storage_file = Path("./storage/composio_oauth.json")

        logger.info("Starting Composio integration initialization...")

        # Load persisted OAuth connections if any
        self._load_persistence()

        # Initialize the Composio toolset
        self._initialize_toolset()

    # [ADDED] Load connections from disk
    def _load_persistence(self):
        if self.storage_file.exists():
            try:
                with self.storage_file.open("r", encoding="utf-8") as f:
                    self._oauth_connections = json.load(f)
                logger.info(f"Loaded Composio OAuth connections from {self.storage_file}")
            except Exception as e:
                logger.warning(f"Error loading Composio OAuth file: {e}")
        else:
            logger.info("No existing Composio OAuth file found.")

    # [ADDED] Save connections to disk
    def _save_persistence(self):
        try:
            self.storage_file.parent.mkdir(exist_ok=True)
            with self.storage_file.open("w", encoding="utf-8") as f:
                json.dump(self._oauth_connections, f, indent=2)
            logger.info("Saved Composio OAuth connections to disk.")
        except Exception as e:
            logger.error(f"Failed to save Composio OAuth connections: {e}")

    def _initialize_toolset(self):
        try:
            api_key = os.environ.get("COMPOSIO_API_KEY")
            if not api_key:
                logger.error("No COMPOSIO_API_KEY in environment")
                return
            self._toolset = ComposioToolSet(
                api_key=api_key,
                entity_id=self._entity_id
            )
            logger.info("Created ComposioToolSet instance")

            # Load the list of apps
            tools = self._toolset.get_tools(actions=["COMPOSIO_LIST_APPS"])
            result = self._toolset.execute_action(
                action="COMPOSIO_LIST_APPS",
                params={},
                entity_id=self._entity_id
            )
            success_value = result.get("success") or result.get("successfull")
            if success_value:
                apps_data = result.get("data", {})
                apps_list = apps_data.get("apps", [])
                for app_info in apps_list:
                    key = app_info.get("key", "").upper()
                    if key:
                        self._available_apps[key] = app_info
                logger.info(f"Fetched {len(self._available_apps)} apps from Composio meta-app")
            else:
                logger.warning("COMPOSIO_LIST_APPS action failed.")
        except Exception as e:
            logger.error(f"Error init Composio: {e}", exc_info=True)
            self._available_apps = {}

    def mark_app_connected(self, app_name: str, connection_id: str):
        """Utility to mark an app as connected in our local _oauth_connections dict."""
        upper_app = app_name.upper()
        self._oauth_connections[upper_app] = {
            "connected": True,
            "connection_id": connection_id
        }
        logger.info(f"mark_app_connected: Marked {upper_app} as connected with connection_id={connection_id}")

        # [ADDED] Persist updated connections to disk
        self._save_persistence()

    async def initiate_oauth_flow(self, app_name: str, redirect_url: str) -> Dict[str, Any]:
        """Begin an OAuth connection for a given app."""
        if not self._toolset:
            return {"success": False, "error": "Toolset not initialized"}

        try:
            upper_app = app_name.upper()
            app_info = self._available_apps.get(upper_app)
            if not app_info:
                return {"success": False, "error": f"Unknown app: {app_name}"}

            connection_req = self._toolset.initiate_connection(
                redirect_url=redirect_url,
                entity_id=self._entity_id,
                app=app_info["key"]  # e.g. "twitter"
            )
            # Some versions of Composio call it 'connectionId', or 'connectedAccountId'
            conn_id = getattr(connection_req, "connectionId", None)
            if not conn_id:
                conn_id = getattr(connection_req, "connectedAccountId", None)
            if not conn_id:
                return {
                    "success": False,
                    "error": "'ConnectionRequestModel' object has no attribute 'connectionId'"
                }

            return {
                "success": True,
                "redirect_url": connection_req.redirectUrl,
                "connection_id": conn_id
            }
        except Exception as e:
            logger.error(f"initiate_oauth_flow error for {app_name}: {e}", exc_info=True)
            return {"success": False, "error": str(e)}

    async def handle_oauth_callback(self, connection_id: str, code: str) -> Dict[str, Any]:
        """
        Finalize the OAuth flow for a given connection_id using the code from the provider.
        Then store 'connected' in _oauth_connections so our front-end can see that it's connected.
        """
        if not self._toolset:
            return {"success": False, "error": "Toolset not initialized"}

        try:
            result = self._toolset.complete_connection(connection_id=connection_id, code=code)
            if result.success:
                # Mark as connected
                app_key = result.app.upper() if result.app else "UNKNOWN"
                self.mark_app_connected(app_key, connection_id)
                logger.info(f"handle_oauth_callback: Marked {app_key} as connected.")
            else:
                logger.warning(f"handle_oauth_callback: success=False for {result.app}")

            return {
                "success": result.success,
                "app": result.app,
                "message": "Connection successful" if result.success else "Connection failed"
            }
        except Exception as e:
            logger.error(f"Error in handle_oauth_callback: {e}", exc_info=True)
            return {"success": False, "error": str(e)}

    def mark_app_connected_without_code(self, app_name: str, connected_account_id: str):
        """
        If Composio doesn't require .complete_connection for some flows 
        but returns connectedAccountId in the callback, 
        we can directly mark the app as connected.
        """
        self.mark_app_connected(app_name, connected_account_id)

    async def list_available_integrations(self) -> List[Dict[str, Any]]:
        """
        Return a list of all apps from _available_apps,
        with "connected" = True if we've tracked them in _oauth_connections.
        """
        results = []
        for key, info in self._available_apps.items():
            upper_key = key.upper()
            is_connected = False
            if upper_key in self._oauth_connections and self._oauth_connections[upper_key].get("connected"):
                is_connected = True

            results.append({
                "name": upper_key,  # e.g. "TWITTER"
                "display_name": info.get("name", upper_key),
                "connected": is_connected,
                "oauth_supported": True,
            })
        return results

    async def list_actions_for_app(self, app_name: str) -> Dict[str, Any]:
        """
        Returns a structure with all possible Composio actions for the given app_name,
        using a direct GET call to Composio's /api/v2/actions/list/all endpoint.

        Example return structure:
        {
            "success": True,
            "actions": [
                "TWITTER_TWEET_CREATE", 
                "TWITTER_DM_SEND", 
                ... 
            ]
        }
        """
        upper_app = app_name.upper()

        # Check if the app is recognized in our local cache
        if upper_app not in self._available_apps:
            return {"success": False, "error": f"App '{app_name}' not recognized in _available_apps"}
        # Check if the app is connected
        if not self._oauth_connections.get(upper_app, {}).get("connected"):
            return {"success": False, "error": f"App '{app_name}' is not connected yet"}

        api_key = os.environ.get("COMPOSIO_API_KEY")
        if not api_key:
            return {"success": False, "error": "No COMPOSIO_API_KEY set in environment"}

        base_url = "https://backend.composio.dev/api/v2/actions/list/all"
        headers = {"x-api-key": api_key}
        params = {"apps": app_name.lower()}  # Composio expects lowercased

        try:
            resp = requests.get(base_url, headers=headers, params=params, timeout=10)
            if resp.status_code == 200:
                data_json = resp.json()
                items = data_json.get("items", [])
                actions = []
                for item in items:
                    action_key = item.get("actionKey")
                    if action_key:
                        actions.append(action_key)
                    else:
                        display_name = item.get("displayName")
                        if display_name:
                            actions.append(display_name)
                return {"success": True, "actions": actions}
            else:
                logger.error(f"Composio API returned {resp.status_code} for app {app_name}")
                return {
                    "success": False,
                    "error": f"Composio returned status {resp.status_code}"
                }
        except Exception as ex:
            logger.error(f"Error retrieving actions for {app_name} from Composio: {ex}", exc_info=True)
            return {"success": False, "error": str(ex)}


# Global single instance
composio_manager = ComposioManager()


##### my_digital_being/framework/main.py #####
import json
import logging
from pathlib import Path
from typing import Dict, Any, Optional
import asyncio
from datetime import datetime

from .memory import Memory
from .state import State
from .activity_selector import ActivitySelector
from .activity_loader import ActivityLoader
from .shared_data import SharedData
from .activity_decorator import ActivityResult

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class DigitalBeing:
    def __init__(self, config_path: Optional[str] = None):
        # Use the config directory relative to this file's location
        if config_path is None:
            config_path = str(Path(__file__).parent.parent / 'config')
        self.config_path = Path(config_path)
        self.configs = self._load_configs()
        self.shared_data = SharedData()
        self.memory = Memory()
        self.state = State()
        self.activity_loader = ActivityLoader()
        self.activity_selector = ActivitySelector(
            self.configs.get("activity_constraints", {}),
            self.state
        )

    def _load_configs(self) -> Dict[str, Any]:
        """Load all configuration files."""
        configs = {}
        config_files = [
            "character_config.json",
            "activity_constraints.json",
            "skills_config.json"
        ]

        for config_file in config_files:
            try:
                with open(self.config_path / config_file, 'r', encoding='utf-8') as f:
                    configs[config_file.replace('.json', '')] = json.load(f)
            except Exception as e:
                logger.error(f"Failed to load config {config_file}: {e}")
                configs[config_file.replace('.json', '')] = {}

        return configs

    def initialize(self):
        """Initialize the digital being."""
        logger.info("Initializing digital being...")

        # Load configurations
        self.configs = self._load_configs()
        logger.info("Configurations loaded")

        # Register API key requirements from skills_config
        skills_config = self.configs.get("skills_config", {})
        from framework.api_management import api_manager  # Avoid top-level import loops

        logger.info("Registering API key requirements for skills...")
        for skill_name, maybe_skill_dict in skills_config.items():
            # SKIP STR KEYS LIKE 'default_llm_skill'
            if not isinstance(maybe_skill_dict, dict):
                logger.debug(f"Skipping non-dict skill config: {skill_name} -> {maybe_skill_dict}")
                continue

            if maybe_skill_dict.get("enabled", False):
                required_keys = maybe_skill_dict.get("required_api_keys", [])
                if required_keys:
                    api_manager.register_required_keys(skill_name, required_keys)
                    logger.info(f"Registered API key requirements for {skill_name}: {required_keys}")

        # Initialize sub-components
        self.memory.initialize()
        self.state.initialize(self.configs.get("character_config", {}))

        # Load activities
        self.activity_loader.load_activities()
        self.shared_data.initialize()

        # Set loader in selector
        self.activity_selector.set_activity_loader(self.activity_loader)

        logger.info("Digital being initialization complete")

    def is_configured(self) -> bool:
        """
        Check if being is 'configured'.
        We look at character_config for 'setup_complete': true
        """
        char_cfg = self.configs.get("character_config", {})
        return bool(char_cfg.get("setup_complete", False))

    async def run(self):
        """
        Main run loop. If not configured, we skip activity selection 
        (but keep looping so the server can remain up).
        """
        logger.info("Starting digital being main loop...")

        try:
            while True:
                # If not configured, skip picking an activity
                if not self.is_configured():
                    logger.warning("Digital Being NOT configured. Skipping activity execution.")
                    await asyncio.sleep(3)
                    continue

                current_activity = self.activity_selector.select_next_activity()
                if current_activity:
                    logger.info(f"Selected activity: {current_activity.__class__.__name__}")
                    await self.execute_activity(current_activity)

                self.state.update()
                self.memory.persist()
                await asyncio.sleep(1)  # short delay to avoid busy-waiting

        except KeyboardInterrupt:
            logger.info("Shutting down digital being...")
            self.cleanup()

    async def execute_activity(self, activity) -> ActivityResult:
        """Execute a selected activity."""
        try:
            logger.info(f"Starting execution of activity: {activity.__class__.__name__}")
            result = await activity.execute(self.shared_data)

            if not isinstance(result, ActivityResult):
                logger.warning(f"Activity {activity.__class__.__name__} did not return an ActivityResult")
                result = ActivityResult(
                    success=bool(result),
                    data=result if result else None,
                    error="Invalid result type" if not result else None
                )

            # Store the activity result
            activity_record = {
                'timestamp': datetime.now().isoformat(),
                'activity_type': activity.__class__.__name__,
                'result': result.to_dict()
            }
            self.memory.store_activity_result(activity_record)

            if result.success:
                logger.info(f"Successfully executed: {activity.__class__.__name__}")
                self.state.record_activity_completion()
            else:
                logger.warning(f"Activity returned failure: {activity.__class__.__name__}")

            return result

        except Exception as e:
            error_msg = f"Failed to execute {activity.__class__.__name__}: {e}"
            logger.error(error_msg)

            error_result = ActivityResult(
                success=False,
                error=str(e)
            )
            self.memory.store_activity_result({
                'timestamp': datetime.now().isoformat(),
                'activity_type': activity.__class__.__name__,
                'result': error_result.to_dict()
            })

            return error_result

    def cleanup(self):
        """Cleanup resources before shutdown."""
        self.memory.persist()
        self.state.save()
        logger.info("Cleanup completed")


if __name__ == "__main__":
    import asyncio
    being = DigitalBeing()
    being.initialize()
    asyncio.run(being.run())


##### my_digital_being/framework/memory.py #####
"""Memory management system for storing and retrieving activity history."""
import json
import logging
from pathlib import Path
from typing import Dict, List, Any
from datetime import datetime, timezone

logger = logging.getLogger(__name__)

class Memory:
    def __init__(self, storage_path: str = "./storage"):
        self.storage_path = Path(storage_path)
        self.storage_path.mkdir(exist_ok=True)
        self.short_term_memory: List[Dict[str, Any]] = []
        self.long_term_memory: Dict[str, Any] = {}
        self.memory_file = self.storage_path / "memory.json"
        self.initialize()

    def initialize(self):
        """Initialize memory system."""
        self._load_memory()

    def _load_memory(self):
        """Load memory from persistent storage."""
        try:
            if self.memory_file.exists():
                with open(self.memory_file, 'r') as f:
                    try:
                        data = json.load(f)
                        if isinstance(data, dict):
                            self.long_term_memory = data.get('long_term', {})
                            self.short_term_memory = data.get('short_term', [])
                        else:
                            logger.warning("Invalid memory file format, resetting memory")
                            self.long_term_memory = {}
                            self.short_term_memory = []
                            self.persist()  # Reset the file with proper format
                    except json.JSONDecodeError as je:
                        logger.error(f"Failed to parse memory file: {je}")
                        # Backup corrupted file
                        backup_path = self.memory_file.with_suffix('.json.bak')
                        self.memory_file.rename(backup_path)
                        logger.info(f"Backed up corrupted memory file to {backup_path}")
                        # Reset memory
                        self.long_term_memory = {}
                        self.short_term_memory = []
                        self.persist()  # Create new file with proper format
        except Exception as e:
            logger.error(f"Failed to load memory: {e}")
            self.long_term_memory = {}
            self.short_term_memory = []

    def store_activity_result(self, activity_record: Dict[str, Any]):
        """Store the result of an activity in memory."""
        try:
            # Ensure we have a valid activity record
            if not isinstance(activity_record, dict):
                logger.error("Invalid activity record format")
                return

            # Extract and validate the result
            result = activity_record.get('result', {})
            if isinstance(result, dict):
                # Store standardized activity record with UTC timestamp
                memory_entry = {
                    'timestamp': datetime.now(timezone.utc).isoformat(),
                    'activity_type': activity_record.get('activity_type', 'Unknown'),
                    'success': result.get('success', False),
                    'error': result.get('error'),
                    'data': result.get('data'),
                    'metadata': result.get('metadata', {})
                }
                self.short_term_memory.append(memory_entry)
                self._consolidate_memory()
                self.persist()  # Persist after each update
                logger.info(f"Stored activity result for {memory_entry['activity_type']}")
            else:
                logger.error(f"Invalid result format in activity record: {result}")

        except Exception as e:
            logger.error(f"Failed to store activity result: {e}")

    def _consolidate_memory(self):
        """Consolidate short-term memory into long-term memory."""
        if len(self.short_term_memory) > 100:  # Keep last 100 activities in short-term
            older_memories = self.short_term_memory[:-50]  # Move older ones to long-term
            self.short_term_memory = self.short_term_memory[-50:]

            for memory in older_memories:
                activity_type = memory['activity_type']
                if activity_type not in self.long_term_memory:
                    self.long_term_memory[activity_type] = []
                self.long_term_memory[activity_type].append(memory)

    def get_recent_activities(self, limit: int = 10, offset: int = 0) -> List[Dict[str, Any]]:
        """Get recent activities from memory with success/failure status."""
        # Sort all activities by timestamp in descending order (most recent first)
        all_activities = sorted(
            self.short_term_memory,
            key=lambda x: x['timestamp'],
            reverse=True
        )

        # Apply pagination
        paginated_activities = all_activities[offset:offset + limit]

        # Format timestamps for display
        return [
            {
                'timestamp': self._format_timestamp(activity['timestamp']),
                'activity_type': activity['activity_type'],
                'success': activity['success'],
                'error': activity.get('error'),
                'data': activity.get('data'),
                'metadata': activity.get('metadata', {})
            }
            for activity in paginated_activities
        ]

    def _format_timestamp(self, timestamp_str: str) -> str:
        """Format ISO timestamp to human-readable format."""
        try:
            dt = datetime.fromisoformat(timestamp_str.replace('Z', '+00:00'))
            return dt.strftime('%Y-%m-%d %H:%M:%S %Z')
        except Exception:
            return timestamp_str

    def get_activity_history(self, activity_type: str) -> List[Dict[str, Any]]:
        """Get history of specific activity type."""
        activities = self.long_term_memory.get(activity_type, [])
        return [
            {
                **activity,
                'timestamp': self._format_timestamp(activity['timestamp'])
            }
            for activity in activities
        ]

    def persist(self):
        """Persist memory to storage."""
        try:
            memory_data = {
                'short_term': self.short_term_memory,
                'long_term': self.long_term_memory
            }

            # Write to a temporary file first
            temp_file = self.memory_file.with_suffix('.json.tmp')
            with open(temp_file, 'w') as f:
                json.dump(memory_data, f, indent=2)

            # Rename temporary file to actual file (atomic operation)
            temp_file.replace(self.memory_file)

        except Exception as e:
            logger.error(f"Failed to persist memory: {e}")

    def clear(self):
        """Clear all memory."""
        self.short_term_memory = []
        self.long_term_memory = {}
        self.persist()

    def get_activity_count(self) -> int:
        """Get total number of activities in memory."""
        return len(self.short_term_memory) + sum(len(activities) for activities in self.long_term_memory.values())

    def get_last_activity_timestamp(self) -> str:
        """Get formatted timestamp of the last activity."""
        if not self.short_term_memory:
            return "No activities recorded"

        last_activity = max(self.short_term_memory, key=lambda x: x['timestamp'])
        return self._format_timestamp(last_activity['timestamp'])

##### my_digital_being/framework/secret_storage.py #####
"""Flexible secret storage system that works across different environments."""
import os
import logging
from typing import Optional, Dict, List
from abc import ABC, abstractmethod
from pathlib import Path
from dotenv import load_dotenv

logger = logging.getLogger(__name__)

class SecretStorageBackend(ABC):
    """Abstract base class for secret storage backends."""

    @abstractmethod
    async def get_secret(self, key: str) -> Optional[str]:
        """Retrieve a secret value by key."""
        pass

    @abstractmethod
    async def set_secret(self, key: str, value: str) -> bool:
        """Store a secret value."""
        pass

    @abstractmethod
    async def list_secrets(self) -> List[str]:
        """List all available secret keys."""
        pass

class EnvFileStorage(SecretStorageBackend):
    """Environment file-based secret storage."""

    def __init__(self, env_path: str = None):
        self.env_path = Path(env_path) if env_path else Path(__file__).parent.parent / '.env'
        load_dotenv(self.env_path)

    async def get_secret(self, key: str) -> Optional[str]:
        """Get secret from .env file."""
        return os.environ.get(key)

    async def set_secret(self, key: str, value: str) -> bool:
        """Set secret in .env file."""
        try:
            # Read existing contents
            env_content = {}
            if self.env_path.exists():
                with open(self.env_path, 'r') as f:
                    for line in f:
                        if '=' in line and not line.startswith('#'):
                            k, v = line.strip().split('=', 1)
                            env_content[k] = v

            # Update or add new key
            env_content[key] = value

            # Write back to file
            with open(self.env_path, 'w') as f:
                for k, v in env_content.items():
                    f.write(f"{k}={v}\n")

            # Update current environment
            os.environ[key] = value
            return True
        except Exception as e:
            logger.error(f"Error setting secret in .env file: {e}")
            return False

    async def list_secrets(self) -> List[str]:
        """List all secrets from .env file."""
        return [key for key in os.environ.keys() if key.endswith('_API_KEY')]

class ReplitSecretStorage(SecretStorageBackend):
    """Replit-specific secret storage implementation."""

    def __init__(self):
        """Initialize with EnvFileStorage as backup."""
        self.env_storage = EnvFileStorage()

    async def get_secret(self, key: str) -> Optional[str]:
        """Get secret from Replit's secure storage."""
        try:
            if 'REPL_ID' in os.environ:
                from replit import db
                return db.get(key) or os.environ.get(key)
            return os.environ.get(key)
        except ImportError:
            logger.warning("Replit module not available, falling back to environment variables")
            return os.environ.get(key)
        except Exception as e:
            logger.error(f"Error getting secret from Replit storage: {e}")
            return os.environ.get(key)

    async def set_secret(self, key: str, value: str) -> bool:
        """Set secret using Replit's secure storage and backup to .env."""
        success = False
        try:
            if 'REPL_ID' in os.environ:
                # Update environment variable immediately
                os.environ[key] = value

                # Store in Replit's db
                from replit import db
                db[key] = value
                success = True

            # Always try to update .env file as backup
            env_success = await self.env_storage.set_secret(key, value)
            return success or env_success

        except ImportError:
            logger.warning("Replit module not available, falling back to .env file")
            return await self.env_storage.set_secret(key, value)
        except Exception as e:
            logger.error(f"Error setting secret in Replit storage: {e}")
            return await self.env_storage.set_secret(key, value)

    async def list_secrets(self) -> List[str]:
        """List all secrets from both Replit and environment."""
        try:
            if 'REPL_ID' in os.environ:
                from replit import db
                replit_keys = [key for key in db.keys() if key.endswith('_API_KEY')]
                env_keys = [key for key in os.environ.keys() if key.endswith('_API_KEY')]
                return list(set(replit_keys + env_keys))
            return [key for key in os.environ.keys() if key.endswith('_API_KEY')]
        except ImportError:
            return [key for key in os.environ.keys() if key.endswith('_API_KEY')]

class SecretManager:
    """Main interface for secret management."""

    def __init__(self):
        """Initialize with appropriate backend based on environment."""
        if 'REPL_ID' in os.environ:
            self.backend = ReplitSecretStorage()
            logger.info("Using Replit secret storage backend")
        else:
            self.backend = EnvFileStorage()
            logger.info("Using .env file secret storage backend")

    async def get_api_key(self, skill_name: str, key_name: str) -> Optional[str]:
        """Get an API key for a specific skill."""
        env_key = f"{skill_name.upper()}_{key_name.upper()}_API_KEY"
        return await self.backend.get_secret(env_key)

    async def set_api_key(self, skill_name: str, key_name: str, value: str) -> bool:
        """Securely store an API key."""
        env_key = f"{skill_name.upper()}_{key_name.upper()}_API_KEY"
        success = await self.backend.set_secret(env_key, value)
        if success:
            # Also set any other skills that use the same key
            try:
                from framework.api_management import api_manager
                all_skills = api_manager.get_required_keys()
                for other_skill, required_keys in all_skills.items():
                    if key_name.upper() in [k.upper() for k in required_keys]:
                        other_env_key = f"{other_skill.upper()}_{key_name.upper()}_API_KEY"
                        await self.backend.set_secret(other_env_key, value)
            except Exception as e:
                logger.error(f"Error propagating API key to other skills: {e}")
        return success

    async def check_api_key_exists(self, skill_name: str, key_name: str) -> bool:
        """Check if an API key exists."""
        env_key = f"{skill_name.upper()}_{key_name.upper()}_API_KEY"
        return bool(await self.backend.get_secret(env_key))

    async def list_configured_keys(self) -> Dict[str, List[str]]:
        """Get all configured API keys grouped by skill."""
        secrets = await self.backend.list_secrets()
        configured_keys = {}

        for secret in secrets:
            if secret.endswith('_API_KEY'):
                parts = secret.split('_')
                if len(parts) >= 3:
                    skill_name = parts[0].lower()
                    key_name = '_'.join(parts[1:-1]).lower()

                    if skill_name not in configured_keys:
                        configured_keys[skill_name] = []
                    configured_keys[skill_name].append(key_name)

        return configured_keys

# Global instance
secret_manager = SecretManager()

##### my_digital_being/framework/shared_data.py #####
import logging
from typing import Dict, Any
from threading import Lock

logger = logging.getLogger(__name__)

class SharedData:
    """Thread-safe shared data storage for activities and skills."""
    
    def __init__(self):
        self._data: Dict[str, Any] = {}
        self._locks: Dict[str, Lock] = {}
        self._global_lock = Lock()

    def initialize(self):
        """Initialize shared data storage."""
        with self._global_lock:
            self._data = {
                'system': {},
                'memory': {},
                'state': {},
                'temp': {}
            }
            for category in self._data:
                self._locks[category] = Lock()

    def get(self, category: str, key: str, default: Any = None) -> Any:
        """Get a value from shared data."""
        if category not in self._data:
            logger.warning(f"Attempting to access invalid category: {category}")
            return default

        with self._locks[category]:
            return self._data[category].get(key, default)

    def set(self, category: str, key: str, value: Any) -> bool:
        """Set a value in shared data."""
        if category not in self._data:
            logger.warning(f"Attempting to write to invalid category: {category}")
            return False

        with self._locks[category]:
            self._data[category][key] = value
        return True

    def update(self, category: str, updates: Dict[str, Any]) -> bool:
        """Update multiple values in a category."""
        if category not in self._data:
            logger.warning(f"Attempting to update invalid category: {category}")
            return False

        with self._locks[category]:
            self._data[category].update(updates)
        return True

    def delete(self, category: str, key: str) -> bool:
        """Delete a value from shared data."""
        if category not in self._data:
            logger.warning(f"Attempting to delete from invalid category: {category}")
            return False

        with self._locks[category]:
            if key in self._data[category]:
                del self._data[category][key]
                return True
        return False

    def clear_category(self, category: str) -> bool:
        """Clear all data in a category."""
        if category not in self._data:
            logger.warning(f"Attempting to clear invalid category: {category}")
            return False

        with self._locks[category]:
            self._data[category].clear()
        return True

    def get_category_data(self, category: str) -> Dict[str, Any]:
        """Get all data in a category."""
        if category not in self._data:
            logger.warning(f"Attempting to access invalid category: {category}")
            return {}

        with self._locks[category]:
            return self._data[category].copy()

    def exists(self, category: str, key: str) -> bool:
        """Check if a key exists in a category."""
        if category not in self._data:
            return False

        with self._locks[category]:
            return key in self._data[category]


##### my_digital_being/framework/skill_config.py #####
"""Secure skill configuration management system + dynamic Composio skills."""
import os
import logging
from typing import Dict, Any, Optional, Set, List

logger = logging.getLogger(__name__)

class SkillConfig:
    """Manages secure configuration for manually coded skills, including API keys."""

    # Class-level storage for tracking API key requirements
    _required_keys: Dict[str, Set[str]] = {}
    _initialized_skills: Set[str] = set()

    def __init__(self, skill_name: str):
        """Initialize skill configuration."""
        self.skill_name = skill_name
        self.config: Dict[str, Any] = {}
        self._load_config()

        if skill_name not in SkillConfig._initialized_skills:
            SkillConfig._initialized_skills.add(skill_name)

    def _load_config(self):
        """Load configuration from environment variables."""
        prefix = f"{self.skill_name.upper()}_"
        for key, value in os.environ.items():
            if key.startswith(prefix):
                config_key = key[len(prefix):].lower()
                self.config[config_key] = value

    def get_api_key(self, key_name: str) -> Optional[str]:
        """
        Safely retrieve an API key from environment variables.
        Raises ValueError if the key is required but not found.
        """
        env_key = f"{self.skill_name.upper()}_{key_name.upper()}_API_KEY"
        api_key = os.environ.get(env_key)

        if not api_key and self._is_key_required(key_name):
            error_msg = f"Required API key '{key_name}' not found for skill '{self.skill_name}'"
            logger.error(error_msg)
            raise ValueError(error_msg)

        return api_key

    def get_config(self, key: str, default: Any = None) -> Any:
        """Get a configuration value."""
        return self.config.get(key, default)

    def _is_key_required(self, key_name: str) -> bool:
        """Check if an API key is required for this skill."""
        return (self.skill_name in SkillConfig._required_keys and
                key_name in SkillConfig._required_keys[self.skill_name])

    @classmethod
    def register_required_keys(cls, skill_name: str, required_keys: List[str]) -> bool:
        """Register required API keys for a manually-coded skill."""
        cls._required_keys[skill_name] = set(required_keys)
        missing_keys = []
        for key in required_keys:
            env_key = f"{skill_name.upper()}_{key.upper()}_API_KEY"
            if not os.environ.get(env_key):
                missing_keys.append(key)
        if missing_keys:
            logger.error(f"Missing required API keys for {skill_name}: {', '.join(missing_keys)}")
            return False
        return True

    @classmethod
    def get_required_keys(cls, skill_name: str = None) -> Dict[str, Set[str]]:
        """Get all required API keys, optionally filtered by skill."""
        if skill_name:
            return {skill_name: cls._required_keys.get(skill_name, set())}
        return cls._required_keys.copy()

    @classmethod
    def verify_skill_keys(cls, skill_name: str) -> tuple[bool, list[str]]:
        """Verify that all required API keys for a skill are available."""
        if skill_name not in cls._required_keys:
            return True, []
        missing_keys = []
        for key in cls._required_keys[skill_name]:
            env_key = f"{skill_name.upper()}_{key.upper()}_API_KEY"
            if not os.environ.get(env_key):
                missing_keys.append(key)
        return len(missing_keys) == 0, missing_keys


#
# BELOW: Our new “DynamicComposioSkills” helper for storing discovered actions as if they were skills
#

class DynamicComposioSkills:
    """
    A registry for "dynamic" skill records discovered from Composio apps/actions.
    Each record might look like:
        {
          "skill_name": "composio_twitter_twitter_tweet_create",
          "enabled": True,
          "required_api_keys": ["COMPOSIO"],   # for example
          "metadata": {
             "composio_app": "TWITTER",
             "composio_action": "TWITTER_TWEET_CREATE"
          }
        }
    """

    # In-memory storage of these dynamic skills
    _dynamic_skills: List[Dict[str, Any]] = []

    @classmethod
    def register_composio_actions(cls, app_name: str, actions: List[str]):
        """
        For each action in `actions`, create a dynamic skill record and store it in _dynamic_skills.
        e.g. skill_name = "composio_{app_name}_{action_id}" (all lowercase)
        """
        for action_id in actions:
            skill_name = f"composio_{app_name.lower()}_{action_id.lower()}"
            skill_record = {
                "skill_name": skill_name,
                "enabled": True,
                # You could decide "required_api_keys": ["COMPOSIO"] or none at all
                "required_api_keys": ["COMPOSIO"],
                "metadata": {
                    "composio_app": app_name.upper(),
                    "composio_action": action_id
                }
            }

            # Avoid duplicates if the user calls this multiple times
            if not any(
                s for s in cls._dynamic_skills
                if s["skill_name"] == skill_name
            ):
                cls._dynamic_skills.append(skill_record)
                logger.info(f"[DynamicComposioSkills] Registered {skill_name}")

    @classmethod
    def get_all_dynamic_skills(cls) -> List[Dict[str, Any]]:
        """Return the entire list of dynamic Composio-based skill records."""
        return cls._dynamic_skills.copy()

    @classmethod
    def find_skill_by_name(cls, skill_name: str) -> Optional[Dict[str, Any]]:
        """Look up a single dynamic skill record by name."""
        for skill in cls._dynamic_skills:
            if skill["skill_name"] == skill_name:
                return skill
        return None


##### my_digital_being/framework/state.py #####
import json
import logging
from pathlib import Path
from typing import Dict, Any
from datetime import datetime

logger = logging.getLogger(__name__)

class State:
    def __init__(self, state_path: str = "./storage"):
        self.state_path = Path(state_path)
        self.state_path.mkdir(exist_ok=True)
        self.state_file = self.state_path / "state.json"
        self.current_state: Dict[str, Any] = {
            'mood': 'neutral',
            'energy': 1.0,
            'last_activity_timestamp': None,
            'active_tasks': [],
            'personality': {}
        }

    def initialize(self, character_config: Dict[str, Any]):
        """Initialize state with character configuration."""
        self._load_state()
        self.current_state['personality'] = character_config.get('personality', {})
        self.save()

    def _load_state(self):
        """Load state from persistent storage."""
        try:
            if self.state_file.exists():
                with open(self.state_file, 'r') as f:
                    self.current_state = json.load(f)
        except Exception as e:
            logger.error(f"Failed to load state: {e}")

    def update(self):
        """Update state based on current conditions."""
        current_time = datetime.now()

        # Update energy levels
        if self.current_state['last_activity_timestamp']:
            last_activity = datetime.fromisoformat(
                self.current_state['last_activity_timestamp']
            )
            time_diff = (current_time - last_activity).total_seconds()
            self.current_state['energy'] = min(
                1.0,
                self.current_state['energy'] + (time_diff / 3600) * 0.1
            )

        # Only update timestamp if there was a successful activity completion
        if hasattr(self, '_last_completed_activity'):
            self.current_state['last_activity_timestamp'] = current_time.isoformat()
            delattr(self, '_last_completed_activity')

        self.save()

    def get_current_state(self) -> Dict[str, Any]:
        """Get current state."""
        return self.current_state.copy()

    def update_mood(self, new_mood: str):
        """Update the current mood."""
        self.current_state['mood'] = new_mood
        self.save()

    def consume_energy(self, amount: float):
        """Consume energy for an activity."""
        self.current_state['energy'] = max(0.0, self.current_state['energy'] - amount)
        self.save()

    def record_activity_completion(self):
        """Mark that an activity was completed successfully."""
        self._last_completed_activity = True
        self.save()

    def add_active_task(self, task_id: str):
        """Add an active task."""
        if task_id not in self.current_state['active_tasks']:
            self.current_state['active_tasks'].append(task_id)
            self.save()

    def remove_active_task(self, task_id: str):
        """Remove an active task."""
        if task_id in self.current_state['active_tasks']:
            self.current_state['active_tasks'].remove(task_id)
            self.save()

    def save(self):
        """Save current state to persistent storage."""
        try:
            with open(self.state_file, 'w') as f:
                json.dump(self.current_state, f, indent=2)
        except Exception as e:
            logger.error(f"Failed to save state: {e}")

##### my_digital_being/skills/skill_chat.py #####
"""OpenAI Chat Completion skill."""
import os
import logging
from typing import Optional, Dict, Any
from openai import OpenAI, APIError
from framework.api_management import api_manager

logger = logging.getLogger(__name__)

class ChatSkill:
    """Skill for chat completions using OpenAI's API."""

    def __init__(self):
        """Initialize the chat skill."""
        self.client: Optional[OpenAI] = None
        self.skill_name = "openai_chat"
        self.required_api_keys = ["openai"]
        # Register API key requirements
        api_manager.register_required_keys(self.skill_name, self.required_api_keys)

    async def initialize(self) -> bool:
        """Initialize the OpenAI client."""
        try:
            api_key = await api_manager.get_api_key(self.skill_name, "openai")
            if not api_key:
                logger.error("OpenAI API key not configured")
                return False

            self.client = OpenAI(api_key=api_key)
            # Test the API key with a minimal request
            response = self.client.chat.completions.create(
                model="gpt-4o",  # the newest OpenAI model is "gpt-4o" which was released May 13, 2024
                messages=[{"role": "user", "content": "Test"}],
                max_tokens=1
            )
            return bool(response)
        except Exception as e:
            logger.error(f"Failed to initialize chat skill: {e}")
            return False

    async def get_chat_completion(self, prompt: str, 
                                system_prompt: str = "You are a helpful AI assistant.", 
                                max_tokens: int = 150) -> Dict[str, Any]:
        """
        Get a chat completion response.

        Args:
            prompt: The user's input prompt
            system_prompt: Optional system message to set the AI's behavior
            max_tokens: Maximum tokens in the response

        Returns:
            Dictionary containing success status, response data or error
        """
        if not self.client:
            return {
                "success": False,
                "error": "Chat skill not initialized",
                "data": None
            }

        try:
            response = self.client.chat.completions.create(
                model="gpt-4o",  # the newest OpenAI model is "gpt-4o" which was released May 13, 2024
                messages=[
                    {"role": "system", "content": system_prompt},
                    {"role": "user", "content": prompt}
                ],
                max_tokens=max_tokens,
                temperature=0.7
            )

            return {
                "success": True,
                "data": {
                    "content": response.choices[0].message.content,
                    "finish_reason": response.choices[0].finish_reason,
                    "model": response.model,
                },
                "error": None
            }

        except APIError as e:
            logger.error(f"OpenAI API error: {e}")
            return {
                "success": False,
                "error": str(e),
                "data": None
            }
        except Exception as e:
            logger.error(f"Unexpected error in chat completion: {e}")
            return {
                "success": False,
                "error": f"Unexpected error: {str(e)}",
                "data": None
            }

# Global instance
chat_skill = ChatSkill()

##### my_digital_being/skills/skill_generate_image.py #####
"""Image generation skill implementation."""
import logging
from typing import Dict, Any, Tuple
import random
from framework.api_management import api_manager

logger = logging.getLogger(__name__)

class ImageGenerationSkill:
    def __init__(self, config: Dict[str, Any]):
        """Initialize the image generation skill with secure API key handling."""
        self.enabled = config.get('enabled', False)
        self.max_generations = config.get('max_generations_per_day', 50)
        self.supported_formats = config.get('supported_formats', ['png', 'jpg'])
        self.generations_count = 0

        # Register required API keys
        api_manager.register_required_keys("image_generation", ["OPENAI"])

    async def can_generate(self) -> bool:
        """Check if image generation is allowed."""
        if not self.enabled:
            logger.warning("Image generation is disabled")
            return False

        if self.generations_count >= self.max_generations:
            logger.warning("Daily generation limit reached")
            return False

        # Verify API key exists and is configured
        api_key = await api_manager.check_api_key_exists("image_generation", "OPENAI")
        if not api_key:
            logger.error("OpenAI API key not configured for image generation")
            return False

        return True

    async def generate_image(
        self,
        prompt: str,
        size: Tuple[int, int] = (512, 512),
        format: str = 'png'
    ) -> Dict[str, Any]:
        """Generate an image based on the prompt."""
        if not await self.can_generate():
            error_msg = "Image generation is not available (disabled, limit reached, or not configured)"
            logger.error(error_msg)
            return {
                'success': False,
                'error': error_msg
            }

        if format not in self.supported_formats:
            error_msg = f'Unsupported format. Use: {self.supported_formats}'
            logger.error(error_msg)
            return {
                'success': False,
                'error': error_msg
            }

        try:
            # Get API key from api_manager
            api_key = await api_manager.get_api_key("image_generation", "OPENAI")
            if not api_key:
                error_msg = 'OpenAI API key not configured'
                logger.error(error_msg)
                return {
                    'success': False,
                    'error': error_msg
                }

            # In a real implementation, this would use the API key to call the image generation service
            logger.info(f"Generating image for prompt: {prompt}")

            # Increment counter only on successful generation
            self.generations_count += 1

            # Simulate generation result
            image_data = {
                'width': size[0],
                'height': size[1],
                'format': format,
                'seed': random.randint(1000, 9999),
                'generation_id': f"gen_{self.generations_count}"
            }

            return {
                'success': True,
                'image_data': image_data,
                'metadata': {
                    'prompt': prompt,
                    'generation_number': self.generations_count
                }
            }
        except Exception as e:
            logger.error(f"Failed to generate image: {e}")
            return {
                'success': False,
                'error': str(e)
            }

    def reset_counts(self):
        """Reset the generation counter."""
        self.generations_count = 0

##### my_digital_being/skills/skill_lite_llm.py #####


"""
LiteLLM Skill
Allows usage of multiple providers (Anthropic, OpenAI, OpenRouter, etc.) via litellm.
We read 'model_name' from skills_config["lite_llm"] if present,
and optionally fetch an API key from the secret manager for "LITELLM".
"""

import os
import logging
from typing import Optional, Dict, Any
from framework.api_management import api_manager
from framework.skill_config import SkillConfig

# litellm
from litellm import completion

logger = logging.getLogger(__name__)

class LiteLLMSkill:
    """Skill for chat/completion using LiteLLM."""

    def __init__(self):
        self.skill_name = "lite_llm"
        self.required_api_keys = ["LITELLM"]
        api_manager.register_required_keys(self.skill_name, self.required_api_keys)

        self.config = SkillConfig(self.skill_name)
        self.model_name: Optional[str] = None
        self._initialized = False

    async def initialize(self) -> bool:
        try:
            # Pull model_name from skill config
            self.model_name = self.config.get_config("model_name", "openai/gpt-4o")

            # Attempt to fetch the key from secret manager
            key = await api_manager.get_api_key(self.skill_name, "LITELLM")
            if key:
                os.environ["LITELLM_API_KEY"] = key
                logger.info("LiteLLM API key loaded from secret manager.")
            else:
                logger.info("No LITELLM API key found; assuming environment or no key needed.")

            self._initialized = True
            return True

        except Exception as e:
            logger.error(f"Failed to initialize LiteLLMSkill: {e}", exc_info=True)
            self._initialized = False
            return False

    async def get_chat_completion(
        self,
        prompt: str,
        system_prompt: str = "You are a helpful AI assistant.",
        max_tokens: int = 150
    ) -> Dict[str, Any]:
        if not self._initialized:
            return {
                "success": False,
                "error": "LiteLLMSkill not initialized",
                "data": None
            }

        try:
            messages = []
            if system_prompt:
                messages.append({"role": "system", "content": system_prompt})
            messages.append({"role": "user", "content": prompt})

            response = completion(
                model=self.model_name,
                messages=messages,
                max_tokens=max_tokens,
                temperature=0.7
            )
            choices = response.get("choices", [])
            if not choices:
                return {
                    "success": False,
                    "error": "No choices returned from LiteLLM",
                    "data": None
                }

            content = choices[0]["message"]["content"]
            finish_reason = choices[0].get("finish_reason", "N/A")
            used_model = response.get("model", self.model_name)

            return {
                "success": True,
                "data": {
                    "content": content,
                    "finish_reason": finish_reason,
                    "model": used_model,
                },
                "error": None
            }

        except Exception as e:
            logger.error(f"Error in LiteLLMSkill completion: {e}", exc_info=True)
            return {
                "success": False,
                "error": f"LiteLLMSkill error: {e}",
                "data": None
            }

# Global instance
lite_llm_skill = LiteLLMSkill()


##### my_digital_being/skills/skill_web_scraping.py #####
"""
Web Scraping Skill
Uses requests + BeautifulSoup to scrape and parse web content.
No external API keys are required.
"""

import logging
from typing import Optional, List, Dict, Any
import requests
from bs4 import BeautifulSoup
from framework.api_management import api_manager  # For consistency, though no keys are used

logger = logging.getLogger(__name__)

class WebScrapingSkill:
    """
    Skill for basic web scraping using requests and BeautifulSoup.
    No API key required.
    """

    def __init__(self):
        self.skill_name = "web_scraping"
        # This skill does not require API keys, but we can still register it
        # with api_manager if we want consistent skill management
        api_manager.register_required_keys(self.skill_name, [])

    async def scrape(self, url: str, parse: bool = True) -> Optional[Dict[str, Any]]:
        """
        Fetch the HTML content from a URL. If parse=True, parse HTML with BeautifulSoup 
        and return a structured representation.

        Returns:
            Dict with keys:
              - 'status_code': HTTP status
              - 'content': raw HTML
              - 'parsed': optional parse result (if parse=True)
            or None if error
        """
        try:
            logger.info(f"Scraping URL: {url}")
            resp = requests.get(url, timeout=10)
            resp.raise_for_status()

            result = {
                'status_code': resp.status_code,
                'content': resp.text
            }

            if parse:
                soup = BeautifulSoup(resp.text, "html.parser")
                # You could add logic to extract links, headings, etc. 
                # For now, we just store the "soup" as a string or partial structure
                # But you can store it as you like
                result['parsed'] = {
                    'title': soup.title.string if soup.title else None,
                    'body_text': soup.get_text(strip=True)[0:500]  # example snippet
                }

            return result
        except Exception as e:
            logger.error(f"Error scraping {url}: {e}", exc_info=True)
            return None


##### my_digital_being/skills/skill_x_api.py #####
"""X (Twitter) API integration skill."""
import os
import logging
from typing import Dict, Any, Optional
import requests
from requests_oauthlib import OAuth1Session
from framework.skill_config import SkillConfig
from framework.api_management import api_manager

logger = logging.getLogger(__name__)

class XAPIError(Exception):
    """Custom exception for X API errors"""
    pass

class XAPISkill:
    """Skill for interacting with X (Twitter) API."""

    def __init__(self, config: Dict[str, Any]):
        """Initialize skill configuration."""
        self.config = config
        self.enabled = config.get('enabled', False)
        self.rate_limit = config.get('rate_limit', 100)
        self.cooldown_period = config.get('cooldown_period', 300)
        self.posts_count = 0
        self.skill_config = SkillConfig("twitter_posting")
        self.oauth_session: Optional[OAuth1Session] = None

    async def initialize(self) -> bool:
        """Initialize the X API skill with required credentials."""
        try:
            # Register required API keys
            required_keys = [
                "API_KEY",
                "API_SECRET",
                "ACCESS_TOKEN",
                "ACCESS_TOKEN_SECRET"
            ]
            api_manager.register_required_keys("twitter_posting", required_keys)

            # Check for missing credentials
            missing_keys = []
            for key in required_keys:
                if not self.skill_config.get_api_key(key):
                    missing_keys.append(key)

            if missing_keys:
                logger.info(f"Missing X API credentials: {missing_keys}")
                return False  # Let the front-end handle credential requests

            # Try to authenticate if we have all credentials
            return await self.authenticate()

        except Exception as e:
            logger.error(f"Failed to initialize X API skill: {e}")
            return False

    def can_post(self) -> bool:
        """Check if posting is allowed based on rate limits."""
        return self.enabled and self.posts_count < self.rate_limit

    async def authenticate(self) -> bool:
        """Set up OAuth session for X API."""
        try:
            api_key = self.skill_config.get_api_key("API_KEY")
            api_secret = self.skill_config.get_api_key("API_SECRET")
            access_token = self.skill_config.get_api_key("ACCESS_TOKEN")
            access_token_secret = self.skill_config.get_api_key("ACCESS_TOKEN_SECRET")

            if not all([api_key, api_secret, access_token, access_token_secret]):
                logger.error("Missing required X API credentials")
                return False

            self.oauth_session = OAuth1Session(
                client_key=api_key,
                client_secret=api_secret,
                resource_owner_key=access_token,
                resource_owner_secret=access_token_secret,
            )
            return True

        except Exception as e:
            logger.error(f"Authentication failed: {e}")
            return False

    async def post_tweet(self, text: str, media_path: Optional[str] = None) -> Dict[str, Any]:
        """Post a tweet with optional media attachment."""
        if not self.can_post():
            return {
                "success": False,
                "error": "Rate limit exceeded or skill disabled"
            }

        if not self.oauth_session:
            if not await self.authenticate():
                return {"success": False, "error": "Authentication failed"}

        try:
            # Handle media upload if provided
            media_id = None
            if media_path and os.path.exists(media_path):
                media_id = await self._upload_media(media_path)

            # Prepare tweet payload
            post_payload = {"text": text}
            if media_id:
                post_payload["media"] = {"media_ids": [media_id]}

            # Post tweet
            response = self.oauth_session.post(
                "https://api.twitter.com/2/tweets",
                json=post_payload
            )

            if response.status_code != 201:
                error_data = response.json() if response.text else {}
                raise XAPIError(f"Failed to post tweet: {error_data}")

            self.posts_count += 1
            return {
                "success": True,
                "tweet_id": response.json()["data"]["id"],
                "content": text
            }

        except Exception as e:
            logger.error(f"Failed to post tweet: {e}")
            return {"success": False, "error": str(e)}

    async def _upload_media(self, media_path: str) -> Optional[str]:
        """Upload media to X and return media_id."""
        try:
            with open(media_path, 'rb') as f:
                files = {"media": f}
                upload_response = self.oauth_session.post(
                    "https://upload.twitter.com/1.1/media/upload.json",
                    files=files
                )

            if upload_response.status_code != 200:
                logger.error(f"Failed to upload media. Status code: {upload_response.status_code}")
                return None

            media_data = upload_response.json()
            return media_data.get("media_id_string")

        except Exception as e:
            logger.error(f"Media upload failed: {e}")
            return None

    def reset_counts(self):
        """Reset the post counter."""
        self.posts_count = 0

##### my_digital_being/tools/onboard.py #####
import os
import json
import logging
import sys
import asyncio
from pathlib import Path

# We'll need to call api_manager in a synchronous context
from framework.api_management import api_manager
from framework.activity_loader import ActivityLoader  # [ADDED] for dynamically listing activities

# Adjust these if your config is stored differently:
CHARACTER_CONFIG_FILE = Path(__file__).parent.parent / "config" / "character_config.json"
SKILLS_CONFIG_FILE = Path(__file__).parent.parent / "config" / "skills_config.json"
ACTIVITY_CONSTRAINTS_FILE = Path(__file__).parent.parent / "config" / "activity_constraints.json"

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def load_json_config(path: Path) -> dict:
    """Helper to safely load JSON config from a file."""
    if not path.exists():
        return {}
    try:
        with open(path, 'r', encoding='utf-8') as f:
            return json.load(f)
    except Exception as e:
        logger.error(f"Failed to load {path.name}: {e}")
        return {}

def save_json_config(path: Path, data: dict) -> None:
    """Helper to save JSON config atomically."""
    temp_file = path.with_suffix('.tmp')
    try:
        with open(temp_file, 'w', encoding='utf-8') as f:
            json.dump(data, f, indent=2)
        temp_file.replace(path)
    except Exception as e:
        logger.error(f"Failed to save {path.name}: {e}")

def prompt_user(prompt_text: str, default: str = None) -> str:
    """Prompt user for input with an optional default."""
    if default is not None:
        user_input = input(f"{prompt_text} [{default}]: ").strip()
        return user_input if user_input else default
    else:
        return input(f"{prompt_text}: ").strip()

def prompt_yes_no(question: str, default: str = "yes") -> bool:
    """
    Prompt user for a yes/no answer with a default.
    Returns True if 'yes', False if 'no'.
    """
    yes_answers = ["yes", "y"]
    no_answers = ["no", "n"]

    if default.lower() in yes_answers:
        prompt_str = f"{question} [Y/n]: "
    else:
        prompt_str = f"{question} [y/N]: "

    while True:
        choice = input(prompt_str).strip().lower()
        if choice == "" and default:
            choice = default.lower()
        if choice in yes_answers:
            return True
        if choice in no_answers:
            return False
        print("Please respond with 'y' or 'n'.")


#
# Helper to set an API key synchronously by calling the async function
#
def set_api_key_sync(skill_name: str, key_name: str, value: str) -> bool:
    """Call api_manager.set_api_key(...) in a blocking manner for CLI convenience."""
    try:
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        result = loop.run_until_complete(api_manager.set_api_key(skill_name, key_name, value))
        loop.close()
        return bool(result.get("success", False))
    except Exception as e:
        logger.error(f"Error setting API key for {skill_name} -> {key_name}: {e}")
        return False

def configure_litellm(skills_config: dict) -> None:
    """
    Prompt user to configure a 'lite_llm' skill or skip.
    We allow a custom model_name (like 'anthropic/claude-3', 'openrouter/openai/gpt-4', etc.)
    If the user provides an API key, we store it via secret manager, not in skills_config.
    """
    print("\n--- LiteLLM Configuration ---")
    if prompt_yes_no("Would you like to configure LiteLLM (supports Anthropic, OpenAI, XAI, OpenRouter, etc.)?", "yes"):
        if "lite_llm" not in skills_config:
            skills_config["lite_llm"] = {
                "enabled": True,
                "required_api_keys": ["LITELLM"],  # We'll define "LITELLM" as the key name
                "api_key_mapping": {
                    "LITELLM": "LITELLM_API_KEY"
                },
                "model_name": None
            }
        else:
            # Ensure it's enabled
            skills_config["lite_llm"]["enabled"] = True
            rkeys = skills_config["lite_llm"].setdefault("required_api_keys", [])
            if "LITELLM" not in rkeys:
                rkeys.append("LITELLM")
            amap = skills_config["lite_llm"].setdefault("api_key_mapping", {})
            if "LITELLM" not in amap:
                amap["LITELLM"] = "LITELLM_API_KEY"

        model_name = prompt_user("Enter model name (e.g. 'anthropic/claude-3' or 'openrouter/openai/gpt-4')", "openai/gpt-4o")
        skills_config["lite_llm"]["model_name"] = model_name

        if prompt_yes_no("Do you want to provide an API key now?", "no"):
            the_key = prompt_user("Enter your LiteLLM-supported API key (or skip)", "")
            if the_key:
                success = set_api_key_sync("lite_llm", "LITELLM", the_key)
                if success:
                    print("LiteLLM API key stored securely!")
                else:
                    print("Failed to store LiteLLM API key. Check logs.")

        use_as_default = prompt_yes_no("Use this lite_llm skill as your default LLM for code generation?", "yes")
        if use_as_default:
            skills_config["default_llm_skill"] = "lite_llm"
    else:
        print("Skipping LiteLLM setup. You can still configure another LLM skill or skip LLM altogether.")

def configure_openai_chat(skills_config: dict) -> None:
    """
    Prompt user to configure openai_chat skill (like GPT).
    If the user provides an API key, we store it in secret manager, not in skills_config.
    """
    print("\n--- OpenAI Chat Configuration ---")
    if "openai_chat" not in skills_config:
        skills_config["openai_chat"] = {
            "enabled": True,
            "required_api_keys": ["OPENAI"],
            "api_key_mapping": {"OPENAI": "OPENAI_API_KEY"}
        }
    else:
        skills_config["openai_chat"]["enabled"] = True

    openai_key = prompt_user("Enter your OPENAI_API_KEY (leave blank if stored in .env or skipping)", "")
    if openai_key:
        success = set_api_key_sync("openai_chat", "OPENAI", openai_key)
        if success:
            print("OpenAI API key stored securely!")
        else:
            print("Failed to store OpenAI key. Check logs.")

    if prompt_yes_no("Use openai_chat as the default LLM skill for code generation?", "no"):
        skills_config["default_llm_skill"] = "openai_chat"

def configure_primary_llm(skills_config: dict) -> None:
    """
    Let user pick from:
    1) LiteLLM skill
    2) OpenAI Chat skill
    3) No LLM
    """
    print("\n--- Primary LLM Choice ---")
    print("1) LiteLLM (anthropic, openai, openrouter, etc.)")
    print("2) OpenAI Chat skill only")
    print("3) None / Skip LLM entirely")

    choice = prompt_user("Enter 1, 2, or 3", default="1")
    if choice == "1":
        configure_litellm(skills_config)
    elif choice == "2":
        configure_openai_chat(skills_config)
    else:
        print("Skipping LLM entirely. No GPT-based code generation or advanced tasks.")
        if "default_llm_skill" in skills_config:
            del skills_config["default_llm_skill"]

def configure_character_basics(character_config: dict) -> None:
    print("\n--- Character Basic Setup ---")
    current_name = character_config.get("name", "Digital Being")
    new_name = prompt_user("Character Name", default=current_name)
    character_config["name"] = new_name

    current_objective = character_config.get("objectives", {}).get("primary", "Assist users")
    new_objective = prompt_user("Primary Objective", default=current_objective)

    if "objectives" not in character_config:
        character_config["objectives"] = {}
    character_config["objectives"]["primary"] = new_objective

def configure_advanced_text(character_config: dict, activity_constraints: dict) -> None:
    if prompt_yes_no("Would you like to define advanced objective text / constraints / examples?", "no"):
        print("\n--- Advanced Objectives ---")
        lines = []
        first_line = prompt_user("Enter multi-line advanced objectives. Press Enter on blank line to finish:", "")
        if first_line.strip():
            lines.append(first_line)
        while True:
            line = input()
            if not line.strip():
                break
            lines.append(line)
        combined = "\n".join(lines)
        if combined:
            character_config.setdefault("objectives", {})
            character_config["objectives"]["advanced"] = combined

        print("\n--- Example Activities ---")
        lines2 = []
        first_line = prompt_user("Enter multi-line example activities. Press Enter on blank line to finish:", "")
        if first_line.strip():
            lines2.append(first_line)
        while True:
            line = input()
            if not line.strip():
                break
            lines2.append(line)
        combined2 = "\n".join(lines2)
        if combined2:
            character_config["example_activities"] = combined2

        print("\n--- General Constraints ---")
        lines3 = []
        first_line = prompt_user("Enter multi-line constraints. Press Enter on blank line to finish:", "")
        if first_line.strip():
            lines3.append(first_line)
        while True:
            line = input()
            if not line.strip():
                break
            lines3.append(line)
        combined3 = "\n".join(lines3)
        if combined3:
            activity_constraints["global_constraints"] = combined3

def configure_other_skills(skills_config: dict) -> None:
    print("\n--- Additional Skills ---")
    skill_names = sorted(skills_config.keys())
    for skill_name in skill_names:
        if skill_name in ["openai_chat", "lite_llm", "default_llm_skill"]:
            continue
        skill_data = skills_config[skill_name]
        is_enabled = skill_data.get("enabled", False)
        user_enable = prompt_yes_no(f"Enable skill '{skill_name}'?", "yes" if is_enabled else "no")
        skill_data["enabled"] = user_enable

        if user_enable and skill_data.get("required_api_keys"):
            for required_key in skill_data["required_api_keys"]:
                env_key = skill_data.get("api_key_mapping", {}).get(required_key, f"{skill_name.upper()}_{required_key}")
                val = prompt_user(f"Enter value for {env_key} (leave blank to skip)", "")
                if val:
                    success = set_api_key_sync(skill_name, required_key, val)
                    if success:
                        print(f"API key for {skill_name}:{required_key} stored!")
                    else:
                        print(f"Failed to store API key for {skill_name}:{required_key}")

# [ADDED] Let user pick which activities to enable/disable (CLI approach)
def configure_activities_cli(activities_config: dict) -> None:
    """
    We discover the available activity classes from the ActivityLoader and let the user
    choose which to enable. We store "enabled": bool in the final JSON under activities_config.
    """
    print("\n--- Activity Enable/Disable Setup (CLI) ---")
    loader = ActivityLoader()
    loader.load_activities()
    found_activities = loader.get_all_activities()  # e.g. {"activity_draw": DrawActivity, ...}

    for mod_name, cls in found_activities.items():
        class_name = cls.__name__  # e.g. "DrawActivity"
        # If we already have an entry, use it; otherwise default to True
        current_enabled = activities_config.get(class_name, {}).get("enabled", True)
        user_enable = prompt_yes_no(f"Enable activity '{class_name}'?", "yes" if current_enabled else "no")

        if class_name not in activities_config:
            activities_config[class_name] = {}
        activities_config[class_name]["enabled"] = user_enable


def main():
    print("=========================================================")
    print(" Welcome to the Autonomous Being CLI Onboarding")
    print("=========================================================")

    character_config_path = CHARACTER_CONFIG_FILE
    skills_config_path = SKILLS_CONFIG_FILE
    activity_constraints_path = ACTIVITY_CONSTRAINTS_FILE

    character_config = load_json_config(character_config_path)
    skills_config = load_json_config(skills_config_path)
    activity_constraints = load_json_config(activity_constraints_path)

    # 1) Primary LLM choice
    configure_primary_llm(skills_config)

    # 2) Basic character config
    configure_character_basics(character_config)

    # 3) Advanced text (objectives, examples, constraints)
    configure_advanced_text(character_config, activity_constraints)

    # 4) Other skills
    configure_other_skills(skills_config)

    # [ADDED] 5) Let user pick which activities to enable/disable in CLI
    # If you prefer to do this only in front-end, you can skip this step or remove it.
    if "activities_config" not in activity_constraints:
        activity_constraints["activities_config"] = {}
    configure_activities_cli(activity_constraints["activities_config"])

    # Save updated (without storing any user-provided API keys in JSON)
    print("\nSaving updated JSON configs...")
    save_json_config(character_config_path, character_config)
    save_json_config(skills_config_path, skills_config)
    save_json_config(activity_constraints_path, activity_constraints)

    print("\nOnboarding complete!")
    print("You may now run 'python -m framework.main' or 'python -m server.server' to launch the AI being.")
    print("-----------------------------------------------------------")


if __name__ == "__main__":
    main()


##### my_digital_being/activities/activity_analyze_daily.py #####
# activities/activity_analyze_daily.py

import logging
from typing import Any
from framework.activity_decorator import activity, ActivityBase, ActivityResult
from framework.memory import Memory
from skills.skill_chat import chat_skill

logger = logging.getLogger(__name__)

@activity(
    name="AnalyzeDailyActivity",
    energy_cost=0.3,
    cooldown=86400,  # 24 hours
    required_skills=["openai_chat"]
)
class AnalyzeDailyActivity(ActivityBase):
    """
    Activity that reviews the last day's logs from memory and produces a reflection
    or summary, storing that reflection in the memory system for future reference.
    """

    def __init__(self):
        super().__init__()
        self.system_prompt = """You are an AI that helps summarize the events, successes,
        or challenges from the digital being's recent memory. Keep the reflection concise and
        highlight any patterns or potential next steps."""

    async def execute(self, shared_data) -> ActivityResult:
        try:
            logger.info("Starting daily analysis of memory...")

            # 1) Initialize the chat skill
            if not await chat_skill.initialize():
                return ActivityResult(
                    success=False,
                    error="Failed to initialize openai_chat skill"
                )

            # 2) Retrieve the last ~10 memory entries for summarization
            memory_obj: Memory = shared_data.get("system", "memory_ref")  # or pass memory another way
            # If not found, fallback to your framework's global memory reference
            if not memory_obj:
                from framework.main import DigitalBeing
                # Fallback to the global being's memory if you prefer
                # In some setups, you can pass it in shared_data, or fetch it from a global reference
                being = DigitalBeing()
                being.initialize()
                memory_obj = being.memory

            recent_activities = memory_obj.get_recent_activities(limit=10, offset=0)

            # 3) Summarize them with the chat skill
            text_snippets = []
            for act in recent_activities:
                snippet = f"- {act['activity_type']}, success={act['success']}, data={act.get('data')}"
                text_snippets.append(snippet)

            combined_text = "\n".join(text_snippets)
            prompt = f"Here are recent logs:\n{combined_text}\n\nProduce a short daily reflection or summary."

            response = await chat_skill.get_chat_completion(
                prompt=prompt,
                system_prompt=self.system_prompt,
                max_tokens=150
            )
            if not response["success"]:
                return ActivityResult(
                    success=False,
                    error=response["error"]
                )

            # 4) Return the reflection as success
            reflection = response["data"]["content"]
            return ActivityResult(
                success=True,
                data={"reflection": reflection},
                metadata={
                    "model": response["data"]["model"],
                    "finish_reason": response["data"]["finish_reason"]
                }
            )

        except Exception as e:
            logger.error(f"Error in AnalyzeDailyActivity: {e}")
            return ActivityResult(success=False, error=str(e))


##### my_digital_being/activities/activity_analyze_new_commits.py #####
import logging
from typing import Dict, Any, List
from datetime import datetime, timedelta

from framework.activity_decorator import activity, ActivityBase, ActivityResult
from framework.api_management import api_manager
from framework.memory import Memory
from skills.skill_chat import chat_skill

logger = logging.getLogger(__name__)

@activity(
    name="analyze_new_commits",
    energy_cost=0.4,
    cooldown=100,  # e.g. 100 seconds for testing; update as needed (e.g. 86400 for daily)
    required_skills=['github_repo_commits']
)
class AnalyzeNewCommitsActivity(ActivityBase):
    """
    Fetch recent commits from GitHub via Composio's GITHUB_LIST_COMMITS action.
    Filter to commits from the last 72 hours, skip any that were already analyzed,
    and then analyze them all in one chat prompt. Returns the analysis in ActivityResult.
    No manual calls to memory_obj.add_activity/store_activity - the system does that for us.
    """

    def __init__(self):
        super().__init__()
        self.composio_action = "GITHUB_LIST_COMMITS"
        self.github_owner = "yoheinakajima"
        self.github_repo = "pippin-py"
        self.github_branch = "main"
        self.lookback_hours = 144  # or 24, etc.

    async def execute(self, shared_data) -> ActivityResult:
        try:
            logger.info("Starting AnalyzeNewCommitsActivity...")

            # 1) Initialize the chat skill
            if not await chat_skill.initialize():
                return ActivityResult(success=False, error="Failed to initialize chat skill")

            # 2) Retrieve memory reference and known commit SHAs (already analyzed)
            memory_obj = self._get_memory(shared_data)
            known_commit_shas = self._get_known_commit_shas(memory_obj)

            # 3) Fetch commits via Composio
            commits_response = self._list_commits_via_composio()
            if not commits_response["success"]:
                error_msg = commits_response.get("error", "Failed to fetch commits")
                return ActivityResult(success=False, error=error_msg)

            commits_data = commits_response.get("data", {}).get("details", [])
            if not commits_data:
                logger.info("No commits returned from GitHub (empty 'details').")
                return ActivityResult(success=True, data={"message": "No commits found."})

            # 4) Filter: only commits from the last X hours
            now_utc = datetime.utcnow()
            fresh_commits = []
            for c in commits_data:
                sha = c.get("sha")
                commit_date_str = c.get("commit", {}).get("author", {}).get("date")  # e.g. 2025-01-07T08:16:00Z
                if not commit_date_str:
                    continue

                try:
                    commit_datetime = datetime.strptime(commit_date_str, "%Y-%m-%dT%H:%M:%SZ")
                except ValueError:
                    logger.warning(f"Could not parse commit date: {commit_date_str}")
                    continue

                if (now_utc - commit_datetime) <= timedelta(hours=self.lookback_hours):
                    fresh_commits.append(c)
                else:
                    logger.info(f"Skipping commit {sha} older than {self.lookback_hours} hours")

            if not fresh_commits:
                return ActivityResult(success=True, data={"message": f"No commits in the last {self.lookback_hours} hours."})

            # 5) Determine which commits are new (not previously analyzed)
            new_commits = [c for c in fresh_commits if c.get("sha") not in known_commit_shas]
            if not new_commits:
                logger.info("All recent commits were already analyzed.")
                return ActivityResult(success=True, data={"message": "No new commits to analyze."})

            # 6) Build a single chat prompt for all new commits
            prompt_text = self._build_batch_prompt(new_commits)
            logger.info(f"Sending one chat prompt with {len(new_commits)} new commits...")

            chat_response = await chat_skill.get_chat_completion(
                prompt=prompt_text,
                system_prompt="You are a code review assistant. Summarize and analyze the following commits in detail.",
                max_tokens=500  # Adjust as needed
            )
            if not chat_response["success"]:
                return ActivityResult(success=False, error=chat_response["error"])

            # 7) Get the combined analysis text from the chat skill
            combined_analysis = chat_response["data"]["content"].strip()

            # 8) Return success with the combined analysis
            # The activity loader or memory system will store the logs automatically.
            # We do not manually call memory_obj.add_activity or store_activity.
            return ActivityResult(
                success=True,
                data={
                    "analysis": combined_analysis,
                    "new_commit_count": len(new_commits),
                    "commits_analyzed": [c.get("sha") for c in new_commits]
                },
                metadata={
                    "model": chat_response["data"].get("model"),
                    "finish_reason": chat_response["data"].get("finish_reason"),
                    "prompt_used": prompt_text
                }
            )

        except Exception as e:
            logger.error(f"Failed to analyze commits: {e}", exc_info=True)
            return ActivityResult(success=False, error=str(e))

    def _get_memory(self, shared_data) -> Memory:
        """
        Fetch or initialize memory object so we can read known commits from past activities.
        """
        system_data = shared_data.get_category_data("system")
        memory_obj: Memory = system_data.get("memory_ref")

        if not memory_obj:
            from framework.main import DigitalBeing
            being = DigitalBeing()
            being.initialize()
            memory_obj = being.memory

        return memory_obj

    def _get_known_commit_shas(self, memory_obj: Memory, limit: int = 50) -> List[str]:
        """
        Reads from memory which commits were already analyzed previously by this same activity.
        We'll skip re-analyzing them.
        """
        recent_activities = memory_obj.get_recent_activities(limit=limit, offset=0)
        known_shas = set()
        for act in recent_activities:
            # If we see "AnalyzeNewCommitsActivity" with success, parse its data
            if act.get("activity_type") == "AnalyzeNewCommitsActivity" and act.get("success"):
                # We rely on the final data posted in the ActivityResult
                # where "commits_analyzed" is a list of commit SHAs or something similar
                commits_analyzed = act.get("data", {}).get("commits_analyzed", [])
                for sha in commits_analyzed:
                    known_shas.add(sha)
        return list(known_shas)

    def _build_batch_prompt(self, commits: List[dict]) -> str:
        """
        Build a single prompt containing all new commits (sha + message).
        We'll ask the LLM to summarize them one by one, note improvements, etc.
        """
        lines = []
        for c in commits:
            sha = c.get("sha", "unknownSHA")[:7]
            message = c.get("commit", {}).get("message", "(no message)")
            lines.append(f"- SHA {sha}: {message}")

        joined_commits = "\n".join(lines)
        prompt = (
            f"Below is a list of {len(commits)} new commits:\n\n"
            f"{joined_commits}\n\n"
            f"Please provide a concise summary of each commit's changes, any improvements needed, "
            f"and note if there are any broader impacts across these commits. "
            f"Be thorough but concise."
        )
        return prompt

    def _list_commits_via_composio(self) -> Dict[str, Any]:
        """
        Calls Composio's GITHUB_LIST_COMMITS action.
        According to your logs, the relevant commits live under "data" -> "details".
        """
        try:
            from framework.composio_integration import composio_manager
            logger.info(
                f"Listing commits from owner='{self.github_owner}', repo='{self.github_repo}', "
                f"branch='{self.github_branch}' using action='{self.composio_action}'"
            )
            response = composio_manager._toolset.execute_action(
                action=self.composio_action,
                params={
                    "owner": self.github_owner,
                    "repo": self.github_repo,
                    "sha": self.github_branch,
                },
                entity_id="MyDigitalBeing"
            )

            # unify "successfull"/"successful"/"success" -> boolean
            success_val = response.get("success", response.get("successfull"))
            if success_val:
                return {"success": True, "data": response.get("data", {})}
            else:
                return {
                    "success": False,
                    "error": response.get("error", "Unknown or missing success key from Composio")
                }

        except Exception as e:
            logger.error(f"Error listing commits from Composio: {e}", exc_info=True)
            return {"success": False, "error": str(e)}


##### my_digital_being/activities/activity_build_or_update.py #####
import logging
import re
from typing import Dict, Any
from framework.activity_decorator import activity, ActivityBase, ActivityResult
from framework.activity_loader import write_activity_code
from skills.skill_chat import chat_skill

from framework.skill_config import DynamicComposioSkills
from framework.api_management import api_manager
from framework.main import DigitalBeing

logger = logging.getLogger(__name__)

@activity(
    name="BuildOrUpdateActivity",
    energy_cost=0.6,
    cooldown=172800,  # 2 days
    required_skills=["openai_chat"]
)
class BuildOrUpdateActivity(ActivityBase):
    """
    Activity that takes suggestions from memory, calls an LLM to generate or update
    Python activities, and writes them to the 'activities/' directory.

    We make two calls to the openai_chat skill:
      1) Get a short filename (activity_*.py).
      2) Generate the actual code snippet, referencing a sample template.

    The final code must:
      - Include `import logging` and any needed imports from `typing` or `framework`
      - Use the @activity decorator from `framework.activity_decorator`
      - Inherit from `ActivityBase`
      - Have an `async def execute(self, shared_data) -> ActivityResult:`
      - Possibly reference known manual-coded skills from `skills/skill_*.py` or
        dynamic Composio skills from `framework.api_management.api_manager` (if relevant).
    """

    def __init__(self):
        super().__init__()
        # Updated system prompt with additional guidelines from our experience
        self.system_prompt = (
            "You are an AI coder that converts user suggestions into valid Python activity files.\n"
            "We have certain code/style constraints based on real-world usage:\n\n"

            "# 1) Decorator usage\n"
            "- The file must define exactly one class decorated with `@activity(...)` from `framework.activity_decorator`.\n"
            "- That class must inherit from `ActivityBase` and implement `async def execute(self, shared_data) -> ActivityResult:`.\n\n"

            "# 2) Manual-coded skill usage\n"
            "- If using, for example, the OpenAI chat skill, do:\n"
            "    from skills.skill_chat import chat_skill\n"
            "    if not await chat_skill.initialize():\n"
            "        return ActivityResult.error_result(\"Chat skill not available\")\n"
            "    response = await chat_skill.get_chat_completion(prompt=\"...\")\n"
            "- DO NOT use self.get_skill_instance(...) or skill lookups in shared_data.\n"
            "- DO NOT define new skill constructors inline.\n\n"

            "# 3) Dynamic Composio skill usage\n"
            "- If referencing a Composio skill, import from framework.api_management:\n"
            "    from framework.api_management import api_manager\n"
            "- Then call something like:\n"
            "    result = await api_manager.composio_manager.execute_action(\n"
            "        action=\"TWITTER_TWEET_CREATE\",  # or e.g. 'Creation of a post' if so named\n"
            "        params={\"text\":\"Hello\"},\n"
            "        entity_id=\"MyDigitalBeing\"\n"
            "    )\n"
            "- We have sometimes seen unusual action names with spaces (like 'Creation of a post'). That's okay.\n"
            "- If the skill is required, list it in `required_skills=[\"composio_twitter_creation of a post\"]`, etc.\n\n"

            "# 4) Memory usage\n"
            "- If referencing memory or retrieving recent activities, you can import from 'framework.main' or 'framework.memory'.\n"
            "- Typically, do:\n"
            "     from framework.main import DigitalBeing\n"
            "     being = DigitalBeing()\n"
            "     being.initialize()\n"
            "     mem = being.memory.get_recent_activities(limit=10)\n"
            "- We do not store the skill or memory object in `shared_data` as a permanent reference. It's optional if you want.\n\n"

            "# 5) Common pitfalls\n"
            "- DO NOT reference unknown modules or placeholders like 'some_module'.\n"
            "- DO NOT rely on fallback calls to uninitialized XAPISkill, if you do not intend them.\n"
            "- If a dynamic skill name differs from your listing (like 'composio_twitter_twitter_tweet_create'), we might need EXACT naming.\n\n"

            "# 6) Example of minimal code snippet\n"
            "```python\n"
            "import logging\n"
            "from typing import Dict, Any\n"
            "from framework.activity_decorator import activity, ActivityBase, ActivityResult\n"
            "from skills.skill_chat import chat_skill\n"
            "from framework.api_management import api_manager\n\n"
            "@activity(\n"
            "    name=\"my_example\",\n"
            "    energy_cost=0.5,\n"
            "    cooldown=3600,\n"
            "    required_skills=[\"openai_chat\"]  # or dynamic composio skill name\n"
            ")\n"
            "class MyExampleActivity(ActivityBase):\n"
            "    \"\"\"Short docstring explaining the activity\"\"\"\n"
            "    def __init__(self):\n"
            "        super().__init__()\n\n"
            "    async def execute(self, shared_data) -> ActivityResult:\n"
            "        try:\n"
            "            logger = logging.getLogger(__name__)\n"
            "            logger.info(\"Executing MyExampleActivity\")\n\n"
            "            # e.g. using openai_chat:\n"
            "            if not await chat_skill.initialize():\n"
            "                return ActivityResult.error_result(\"Chat skill not available\")\n"
            "            result = await chat_skill.get_chat_completion(prompt=\"Hello!\")\n\n"
            "            # or dynamic composio skill, e.g.:\n"
            "            # result2 = await api_manager.composio_manager.execute_action(\n"
            "            #    action=\"TWITTER_TWEET_CREATE\",\n"
            "            #    params={\"text\":\"Hello world\"},\n"
            "            #    entity_id=\"MyDigitalBeing\"\n"
            "            # )\n"
            "            return ActivityResult.success_result({\"message\":\"Done\"})\n"
            "        except Exception as e:\n"
            "            return ActivityResult.error_result(str(e))\n"
            "```\n\n"

            "# 7) Summation\n"
            "Given user suggestions and known skill data, produce EXACT code meeting these standards.\n"
            "No triple backticks. Single @activity class only.\n"
        )

    async def execute(self, shared_data) -> ActivityResult:
        try:
            logger.info("Starting BuildOrUpdateActivity...")

            # 1) Initialize chat skill
            if not await chat_skill.initialize():
                return ActivityResult(success=False, error="Failed to initialize openai_chat skill")

            # 2) Access the being + memory
            being = DigitalBeing()
            being.initialize()
            recent_activities = being.memory.get_recent_activities(limit=20)

            # 3) Gather skill info (both manual + dynamic)
            skills_config = being.configs.get("skills_config", {})
            manual_skill_list = []
            for skill_name, skill_info in skills_config.items():
                if isinstance(skill_info, dict):
                    desc = f"Skill: {skill_name}, enabled={skill_info.get('enabled')}"
                    req_keys = skill_info.get('required_api_keys', [])
                    desc += f", required_api_keys={req_keys}"
                    meta = skill_info.get('metadata', {})
                    if meta:
                        desc += f", metadata={meta}"
                    manual_skill_list.append(desc)

            dynamic_skills = DynamicComposioSkills.get_all_dynamic_skills()
            dynamic_skill_list = []
            for ds in dynamic_skills:
                d_name = ds["skill_name"]
                d_enabled = ds.get("enabled", True)
                d_req = ds.get("required_api_keys", [])
                d_meta = ds.get("metadata", {})
                desc = f"DynamicSkill: {d_name}, enabled={d_enabled}, required_api_keys={d_req}, metadata={d_meta}"
                dynamic_skill_list.append(desc)

            all_skills_block = "\n".join(manual_skill_list + dynamic_skill_list)
            if not all_skills_block.strip():
                all_skills_block = "(No known skills found)"

            # 4) Find last suggestions from memory (SuggestNewActivities)
            suggestion_texts = []
            for act in recent_activities:
                if act['activity_type'] == "SuggestNewActivities":
                    data_content = act.get('data', {})
                    if isinstance(data_content, dict) and 'suggestions' in data_content:
                        suggestion_texts.append(data_content['suggestions'])

            if not suggestion_texts:
                return ActivityResult(
                    success=False,
                    error="No recent suggestions in memory; cannot build new activity"
                )
            combined_suggestions = "\n---\n".join(suggestion_texts)

            # ---------------------------------------------------------------------
            # A) FIRST LLM CALL - GET A SHORT FILENAME
            # ---------------------------------------------------------------------
            filename_prompt = (
                f"User Suggestions:\n{combined_suggestions}\n\n"
                f"Known Skills:\n{all_skills_block}\n\n"
                "Propose a short new file name that starts with 'activity_' and ends with '.py'. "
                "Do NOT provide any code, just the file name (no quotes, no backticks)."
            )

            filename_resp = await chat_skill.get_chat_completion(
                prompt=filename_prompt,
                system_prompt=self.system_prompt,
                max_tokens=50
            )
            if not filename_resp["success"]:
                return ActivityResult(success=False, error=filename_resp["error"])

            raw_filename = filename_resp["data"]["content"].strip()
            match_name = re.search(r'(activity_[\w-]+\.py)', raw_filename)
            if match_name:
                filename = match_name.group(1)
            else:
                filename = "activity_new_suggestion.py"

            # ---------------------------------------------------------------------
            # B) SECOND LLM CALL - GET THE FULL CODE
            # ---------------------------------------------------------------------
            code_prompt = (
                f"User Suggestions:\n{combined_suggestions}\n\n"
                f"Known Skills:\n{all_skills_block}\n\n"
                "Below is an example minimal template that shows how we want to reference manual-coded skills "
                "or dynamic composio skills:\n"
                "```python\n"
                "import logging\n"
                "from typing import Dict, Any\n"
                "from framework.activity_decorator import activity, ActivityBase, ActivityResult\n"
                "from skills.skill_chat import chat_skill\n"
                "from framework.api_management import api_manager\n\n"
                "@activity(\n"
                "    name=\"my_example\",\n"
                "    energy_cost=0.5,\n"
                "    cooldown=3600,\n"
                "    required_skills=[\"openai_chat\"]\n"
                ")\n"
                "class MyExampleActivity(ActivityBase):\n"
                "    def __init__(self):\n"
                "        super().__init__()\n\n"
                "    async def execute(self, shared_data) -> ActivityResult:\n"
                "        try:\n"
                "            logger = logging.getLogger(__name__)\n"
                "            logger.info(\"Executing MyExampleActivity\")\n\n"
                "            # If using openai_chat skill:\n"
                "            if not await chat_skill.initialize():\n"
                "                return ActivityResult.error_result(\"Chat skill not available\")\n"
                "            result = await chat_skill.get_chat_completion(prompt=\"Hello!\")\n\n"
                "            # If using dynamic composio skill, e.g. 'composio_twitter_twitter_tweet_create':\n"
                "            #    result2 = await api_manager.composio_manager.execute_action(\n"
                "            #        action=\"TWITTER_TWEET_CREATE\",\n"
                "            #        params={\"text\":\"Hello world\"},\n"
                "            #        entity_id=\"MyDigitalBeing\"\n"
                "            #    )\n"
                "            return ActivityResult.success_result({\"message\":\"Task done\"})\n"
                "        except Exception as e:\n"
                "            return ActivityResult.error_result(str(e))\n"
                "```\n\n"
                f"Now produce a FULL Python file named {filename} with exactly one activity class that meets the instructions:\n"
                "- Single @activity decorator\n"
                "- Inherit from ActivityBase\n"
                "- Has `async def execute(...)`\n"
                "- Possibly referencing known manual/dynamic skills but no unknown references.\n"
                "- DO NOT wrap your code in triple backticks.\n"
            )

            code_resp = await chat_skill.get_chat_completion(
                prompt=code_prompt,
                system_prompt=self.system_prompt,
                max_tokens=1200
            )
            if not code_resp["success"]:
                return ActivityResult(success=False, error=code_resp["error"])

            code_snippet = code_resp["data"]["content"]
            code_snippet = self._clean_code_snippet(code_snippet)

            # ---------------------------------------------------------------------
            # Write to disk + Reload
            # ---------------------------------------------------------------------
            success = write_activity_code(filename, code_snippet)
            if not success:
                return ActivityResult(
                    success=False,
                    error=f"Failed to write {filename} to disk"
                )

            # Reload so the new activity is recognized immediately
            being.activity_loader.reload_activities()

            return ActivityResult(
                success=True,
                data={
                    "filename": filename,
                    "code_snippet": code_snippet
                },
                metadata={
                    "message": "Activity created/updated and reloaded"
                }
            )

        except Exception as e:
            logger.error(f"Error in BuildOrUpdateActivity: {e}", exc_info=True)
            return ActivityResult(success=False, error=str(e))

    def _clean_code_snippet(self, snippet: str) -> str:
        """
        Remove triple-backtick fences (` ```python ` or ` ``` `) from the snippet,
        plus any leading/trailing whitespace.
        """
        snippet = snippet.strip()
        # Remove any leading ```python or ```
        snippet = re.sub(r'^```(?:python)?', '', snippet, flags=re.IGNORECASE).strip()
        # Remove any trailing ```
        snippet = re.sub(r'```$', '', snippet).strip()
        return snippet


##### my_digital_being/activities/activity_daily_thought.py #####
"""Activity for generating daily thoughts using OpenAI."""
import logging
from datetime import timedelta
from framework.activity_decorator import activity, ActivityBase, ActivityResult
from skills.skill_chat import chat_skill

logger = logging.getLogger(__name__)

@activity(
    name="daily_thought",
    energy_cost=0.4,
    cooldown=1800,  # 30 minutes
    required_skills=['openai_chat']
)
class DailyThoughtActivity(ActivityBase):
    """Generates insightful daily thoughts using OpenAI."""

    def __init__(self):
        super().__init__()
        self.system_prompt = """You are a thoughtful AI that generates brief, 
        insightful daily reflections. Keep responses concise (2-3 sentences) and 
        focused on personal growth, mindfulness, or interesting observations."""

    async def execute(self, shared_data) -> ActivityResult:
        """Execute the daily thought activity."""
        try:
            logger.info("Starting daily thought generation")

            # Initialize required skills
            if not await chat_skill.initialize():
                return ActivityResult.error_result("Failed to initialize chat skill")

            # Generate the thought
            result = await chat_skill.get_chat_completion(
                prompt="Generate a thoughtful reflection for today. Focus on personal growth, mindfulness, or an interesting perspective.",
                system_prompt=self.system_prompt,
                max_tokens=100
            )

            if not result["success"]:
                return ActivityResult.error_result(result["error"])

            return ActivityResult.success_result(
                data={
                    "thought": result["data"]["content"]
                },
                metadata={
                    "model": result["data"]["model"],
                    "finish_reason": result["data"]["finish_reason"]
                }
            )

        except Exception as e:
            logger.error(f"Error in daily thought activity: {e}")
            return ActivityResult.error_result(str(e))

##### my_digital_being/activities/activity_draw.py #####
"""Drawing activity implementation."""
import logging
from typing import Dict, Any
from framework.activity_decorator import activity, ActivityBase, ActivityResult
from skills.skill_generate_image import ImageGenerationSkill
from framework.api_management import api_manager

logger = logging.getLogger(__name__)

@activity(
    name="draw",
    energy_cost=0.6,
    cooldown=7200,  # 2 hours
    required_skills=['image_generation']
)
class DrawActivity(ActivityBase):
    def __init__(self):
        super().__init__()
        self.default_size = (512, 512)
        self.default_format = 'png'

    async def execute(self, shared_data) -> ActivityResult:
        """Execute the drawing activity."""
        try:
            logger.info("Starting drawing activity")

            # Initialize the image generation skill with configuration
            image_skill = ImageGenerationSkill({
                'enabled': True,
                'max_generations_per_day': 50,
                'supported_formats': ['png', 'jpg']
            })

            # Verify the skill can generate images
            if not await image_skill.can_generate():
                error_msg = "Image generation is not available at this time"
                logger.error(error_msg)
                return ActivityResult(
                    success=False,
                    error=error_msg
                )

            # Generate drawing prompt
            prompt = self._generate_prompt(shared_data)

            # Generate the image
            result = await image_skill.generate_image(
                prompt=prompt,
                size=self.default_size,
                format=self.default_format
            )

            if result.get('success'):
                # Store the generated image data
                shared_data.set('memory', f"drawing_{result['image_data']['generation_id']}", {
                    'prompt': prompt,
                    'image_data': result['image_data']
                })

                logger.info(f"Successfully generated image for prompt: {prompt}")
                return ActivityResult(
                    success=True,
                    data={
                        'generation_id': result['image_data']['generation_id'],
                        'prompt': prompt,
                        'image_data': result['image_data']
                    },
                    metadata={
                        'size': self.default_size,
                        'format': self.default_format
                    }
                )
            else:
                error_msg = result.get('error', 'Unknown error')
                logger.error(f"Failed to generate image: {error_msg}")
                return ActivityResult(
                    success=False,
                    error=error_msg
                )

        except Exception as e:
            logger.error(f"Failed to generate drawing: {e}")
            return ActivityResult(
                success=False,
                error=str(e)
            )

    def _generate_prompt(self, shared_data) -> str:
        """Generate a drawing prompt based on current state and memory."""
        state = shared_data.get('state', 'current_state', {})
        personality = state.get('personality', {})
        mood = state.get('mood', 'neutral')

        # Base prompts for different moods
        mood_prompts = {
            'happy': "a sunny landscape with vibrant colors",
            'neutral': "a peaceful scene with balanced composition",
            'sad': "a rainy day with muted colors",
        }

        # Get base prompt from mood
        base_prompt = mood_prompts.get(mood, mood_prompts['neutral'])

        # Modify based on personality
        if personality.get('creativity', 0) > 0.7:
            base_prompt += " with surreal elements"
        if personality.get('curiosity', 0) > 0.7:
            base_prompt += " featuring unexpected details"

        return f"Digital artwork of {base_prompt}, digital art style"

##### my_digital_being/activities/activity_evaluate.py #####
# activities/activity_evaluate.py

import logging
from typing import Dict, Any
from framework.activity_decorator import activity, ActivityBase, ActivityResult
from skills.skill_chat import chat_skill

logger = logging.getLogger(__name__)

@activity(
    name="EvaluateActivity",
    energy_cost=0.3,
    cooldown=86400,  # example: 1 day
    required_skills=["openai_chat"]
)
class EvaluateActivity(ActivityBase):
    """
    Activity that attempts to 'simulate' how effective a newly generated activity might be
    or identify potential problems. This is purely an LLM-based guess, not guaranteed accurate.
    """

    def __init__(self):
        super().__init__()
        self.system_prompt = """You are an AI that evaluates the potential effectiveness
        of newly generated Activities. You consider whether the code is likely to run,
        fits the being's objectives, and avoids major pitfalls.
        Provide a short bullet-point analysis.
        """

    async def execute(self, shared_data) -> ActivityResult:
        try:
            logger.info("Starting EvaluateActivity...")

            if not await chat_skill.initialize():
                return ActivityResult(
                    success=False,
                    error="Failed to initialize openai_chat skill"
                )

            # Possibly fetch the last created/updated code from memory
            from framework.main import DigitalBeing
            being = DigitalBeing()
            being.initialize()
            recents = being.memory.get_recent_activities(limit=10)
            code_found = None

            for act in recents:
                if act['activity_type'] == "BuildOrUpdateActivity" and act.get('data', {}):
                    data_content = act['data']
                    if "code_snippet" in data_content:
                        code_found = data_content["code_snippet"]
                        break

            if not code_found:
                return ActivityResult(success=False, error="No newly generated code found to evaluate")

            prompt_text = (
                f"Here is the code for a newly created activity:\n{code_found}\n\n"
                "Evaluate how effective or risky this might be. Provide bullet points. "
                "Focus on alignment with objectives, potential errors, or improvements."
            )

            response = await chat_skill.get_chat_completion(
                prompt=prompt_text,
                system_prompt=self.system_prompt,
                max_tokens=250
            )
            if not response["success"]:
                return ActivityResult(success=False, error=response["error"])

            evaluation = response["data"]["content"]
            return ActivityResult(
                success=True,
                data={"evaluation": evaluation},
                metadata={
                    "model": response["data"]["model"],
                    "finish_reason": response["data"]["finish_reason"]
                }
            )

        except Exception as e:
            logger.error(f"Error in EvaluateActivity: {e}")
            return ActivityResult(success=False, error=str(e))


##### my_digital_being/activities/activity_fetch_news.py #####
"""Activity for fetching news using web scraping."""
import logging
from typing import Dict, Any, List
from framework.activity_decorator import activity, ActivityBase, ActivityResult

logger = logging.getLogger(__name__)

@activity(
    name="fetch_news",
    energy_cost=0.3,
    cooldown=1800,  # 30 minutes
    required_skills=['web_scraping']
)
class FetchNewsActivity(ActivityBase):
    def __init__(self):
        super().__init__()
        self.topics = ["technology", "science", "art"]
        self.max_articles = 5

    async def execute(self, shared_data) -> ActivityResult:
        """Execute the news fetching activity."""
        try:
            logger.info("Starting news fetch activity")

            # Simulate fetching news
            articles = await self._fetch_articles()

            # Store articles in shared data
            shared_data.set('memory', 'latest_news', articles)

            logger.info(f"Successfully fetched {len(articles)} articles")
            return ActivityResult(
                success=True,
                data={
                    'articles': articles,
                    'count': len(articles)
                },
                metadata={
                    'topics': self.topics,
                    'max_articles': self.max_articles
                }
            )

        except Exception as e:
            logger.error(f"Failed to fetch news: {e}")
            return ActivityResult(
                success=False,
                error=str(e)
            )

    async def _fetch_articles(self) -> List[Dict[str, Any]]:
        """Simulate fetching articles."""
        # In a real implementation, this would use web scraping
        articles = []
        for i in range(self.max_articles):
            articles.append({
                'title': f"Simulated Article {i+1}",
                'topic': self.topics[i % len(self.topics)],
                'summary': f"This is a simulated news article about {self.topics[i % len(self.topics)]}",
                'url': f"https://example.com/article_{i+1}"
            })
        return articles


##### my_digital_being/activities/activity_nap.py #####
"""
Activity for taking a "nap" to simulate resting. 
We store a 'napping' state in shared_data, then return success.
"""
import logging
from typing import Dict, Any, List
from framework.activity_decorator import activity, ActivityBase, ActivityResult

logger = logging.getLogger(__name__)

@activity(
    name="nap",
    energy_cost=0,      # A nap might not cost energy, or you can set 0.2, etc.
    cooldown=1800       # e.g. a 30-minute cooldown between naps
)
class NapActivity(ActivityBase):
    def __init__(self):
        super().__init__()
        self.nap_minutes = 15  # length of nap in minutes

    async def execute(self, shared_data) -> ActivityResult:
        """
        Execute the nap activity.
        We'll simulate a 'nap' by logging that we're napping,
        then store a record in shared_data that a nap took place.
        """
        try:
            logger.info(f"Taking a {self.nap_minutes}-minute nap.")
            # You can imagine "await asyncio.sleep(...)" if you wanted a real delay.

            # Store a record in shared_data:
            # e.g., shared_data.set('body_state', 'currently_napping', True)
            # or log the timestamp, etc.
            shared_data.set('body_state', 'nap_info', {
                'last_nap_duration': self.nap_minutes,
                'timestamp': "Just now!"
            })

            logger.info("Nap finished. Feeling refreshed!")
            return ActivityResult(
                success=True,
                data={
                    'nap_minutes': self.nap_minutes
                },
                metadata={
                    'message': "Nap complete",
                }
            )

        except Exception as e:
            logger.error(f"Nap failed: {e}")
            return ActivityResult(
                success=False,
                error=str(e)
            )


##### my_digital_being/activities/activity_post_a_tweet.py #####
import logging
from typing import Dict, Any, List

from framework.activity_decorator import activity, ActivityBase, ActivityResult
from framework.api_management import api_manager
from framework.memory import Memory
from skills.skill_chat import chat_skill

logger = logging.getLogger(__name__)

@activity(
    name="post_a_tweet",
    energy_cost=0.4,
    cooldown=10000,  # 1 hour
    required_skills=['twitter_posting']
)
class PostTweetActivity(ActivityBase):
    """
    Uses a chat skill (OpenAI) to generate tweet text,
    referencing the character's personality from character_config.
    Checks recent tweets in memory to avoid duplication.
    Posts to Twitter via Composio's "Creation of a post" dynamic action.
    """

    def __init__(self):
        super().__init__()
        self.max_length = 280
        # The Composio action name from your logs
        self.composio_action = "TWITTER_CREATION_OF_A_POST"
        # If you know your Twitter username, you can embed it in the link
        # or fetch it dynamically. Otherwise, substitute accordingly:
        self.twitter_username = "YourUserName"

    async def execute(self, shared_data) -> ActivityResult:
        try:
            logger.info("Starting tweet posting activity...")

            # 1) Initialize the chat skill
            if not await chat_skill.initialize():
                return ActivityResult(success=False, error="Failed to initialize chat skill")

            # 2) Gather personality + recent tweets
            character_config = self._get_character_config(shared_data)
            personality_data = character_config.get("personality", {})
            recent_tweets = self._get_recent_tweets(shared_data, limit=10)

            # 3) Generate tweet text with chat skill
            prompt_text = self._build_chat_prompt(personality_data, recent_tweets)
            chat_response = await chat_skill.get_chat_completion(
                prompt=prompt_text,
                system_prompt="You are an AI that composes tweets with the given personality.",
                max_tokens=100
            )
            if not chat_response["success"]:
                return ActivityResult(success=False, error=chat_response["error"])

            tweet_text = chat_response["data"]["content"].strip()
            if len(tweet_text) > self.max_length:
                tweet_text = tweet_text[: self.max_length - 3] + "..."

            # 4) Post the tweet via Composio
            post_result = self._post_tweet_via_composio(tweet_text)
            if not post_result["success"]:
                error_msg = post_result.get("error", "Unknown error posting tweet via Composio")
                logger.error(f"Tweet posting failed: {error_msg}")
                return ActivityResult(success=False, error=error_msg)

            tweet_id = post_result.get("tweet_id")
            tweet_link = (
                f"https://twitter.com/{self.twitter_username}/status/{tweet_id}"
                if tweet_id else None
            )

            # 5) Return success, adding link & prompt in metadata
            logger.info(f"Successfully posted tweet: {tweet_text[:50]}...")
            return ActivityResult(
                success=True,
                data={
                    "tweet_id": tweet_id,
                    "content": tweet_text
                },
                metadata={
                    "length": len(tweet_text),
                    "method": "composio",
                    "model": chat_response["data"].get("model"),
                    "finish_reason": chat_response["data"].get("finish_reason"),
                    "tweet_link": tweet_link,
                    "prompt_used": prompt_text  # <--- includes the full prompt
                }
            )

        except Exception as e:
            logger.error(f"Failed to post tweet: {e}", exc_info=True)
            return ActivityResult(success=False, error=str(e))

    def _get_character_config(self, shared_data) -> Dict[str, Any]:
        """
        Retrieve character_config from SharedData['system'] or re-init the Being if not found.
        """
        system_data = shared_data.get_category_data("system")
        maybe_config = system_data.get("character_config")
        if maybe_config:
            return maybe_config

        # fallback
        from framework.main import DigitalBeing
        being = DigitalBeing()
        being.initialize()
        return being.configs.get("character_config", {})

    def _get_recent_tweets(self, shared_data, limit: int = 10) -> List[str]:
        """
        Fetch the last N tweets posted (activity_type='PostTweetActivity') from memory.
        """
        system_data = shared_data.get_category_data("system")
        memory_obj: Memory = system_data.get("memory_ref")

        if not memory_obj:
            from framework.main import DigitalBeing
            being = DigitalBeing()
            being.initialize()
            memory_obj = being.memory

        recent_activities = memory_obj.get_recent_activities(limit=50, offset=0)
        tweets = []
        for act in recent_activities:
            if act.get("activity_type") == "PostTweetActivity" and act.get("success"):
                tweet_body = act.get("data", {}).get("content", "")
                if tweet_body:
                    tweets.append(tweet_body)

        return tweets[:limit]

    def _build_chat_prompt(self, personality: Dict[str, Any], recent_tweets: List[str]) -> str:
        """
        Construct the user prompt referencing personality + last tweets.
        """
        trait_lines = [f"{t}: {v}" for t, v in personality.items()]
        personality_str = "\n".join(trait_lines)

        if recent_tweets:
            last_tweets_str = "\n".join(f"- {txt}" for txt in recent_tweets)
        else:
            last_tweets_str = "(No recent tweets)"

        return (
            f"Our digital being has these personality traits:\n"
            f"{personality_str}\n\n"
            f"Here are recent tweets:\n"
            f"{last_tweets_str}\n\n"
            f"Write a new short tweet (under 280 chars), consistent with the above, "
            f"but not repeating old tweets. Avoid hashtags or repeated phrases.\n"
        )

    def _post_tweet_via_composio(self, tweet_text: str) -> Dict[str, Any]:
        """
        Post a tweet using the "Creation of a post" Composio action.
        The response returns {'successfull': True, ...}, not 'success'.
        We'll check 'successfull' or fallback if needed.
        """
        try:
            from framework.composio_integration import composio_manager
            logger.info(f"Posting tweet via Composio action='{self.composio_action}', text='{tweet_text[:50]}...'")

            response = composio_manager._toolset.execute_action(
                action=self.composio_action,
                params={"text": tweet_text},
                entity_id="MyDigitalBeing"
            )

            # The actual success key is "successfull" (with 2 Ls)
            success_val = response.get("success", response.get("successfull"))
            if success_val:
                data_section = response.get("data", {})
                nested_data = data_section.get("data", {})
                tweet_id = nested_data.get("id")
                return {"success": True, "tweet_id": tweet_id}
            else:
                return {
                    "success": False,
                    "error": response.get("error", "Unknown or missing success key")
                }

        except Exception as e:
            logger.error(f"Error in Composio tweet post: {e}", exc_info=True)
            return {"success": False, "error": str(e)}


##### my_digital_being/activities/activity_post_recent_memory_tweet.py #####
import logging
from typing import Dict, Any, List

from framework.activity_decorator import activity, ActivityBase, ActivityResult
from framework.api_management import api_manager
from framework.memory import Memory
from skills.skill_chat import chat_skill

logger = logging.getLogger(__name__)

@activity(
    name="post_recent_memories_tweet",
    energy_cost=0.4,
    cooldown=10000,  # e.g. ~2.7 hours for testing (adjust as needed)
    required_skills=['twitter_posting']
)
class PostRecentMemoriesTweetActivity(ActivityBase):
    """
    Pulls recent memory items (up to N), ignoring certain activity types,
    filters out those used in the previous run of this activity,
    references personality + objectives from character_config,
    composes a short tweet via chat skill, and posts it.
    """

    def __init__(self, num_activities_to_fetch: int = 10):
        super().__init__()
        self.max_length = 280
        self.composio_action = "TWITTER_CREATION_OF_A_POST"
        self.twitter_username = "YourUserName"

        # Activity types to ignore in memory results
        self.ignored_activity_types = [
            "PostRecentMemoriesTweetActivity",  # ignore itself
            "PostTweetActivity"
        ]

        # How many recent memory entries to consider
        self.num_activities_to_fetch = num_activities_to_fetch

    async def execute(self, shared_data) -> ActivityResult:
        try:
            logger.info("Starting PostRecentMemoriesTweetActivity...")

            # 1) Initialize chat skill
            if not await chat_skill.initialize():
                return ActivityResult(success=False, error="Failed to initialize chat skill")

            # 2) Load personality + objectives from character config
            character_config = self._get_character_config(shared_data)
            personality_data = character_config.get("personality", {})
            objectives_data = character_config.get("objectives", {})
            # For example: objectives_data might be {"primary": "Spread positivity"}

            # 3) Fetch recent memories, ignoring certain activity types
            recent_memories = self._get_recent_memories(
                shared_data,
                limit=self.num_activities_to_fetch
            )
            if not recent_memories:
                logger.info("No relevant memories found to tweet about.")
                return ActivityResult(success=True, data={"message": "No recent memories to share."})

            # 4) Find which memories we used last time (to avoid repeats)
            used_memories_last_time = self._get_memories_used_last_time(shared_data)
            logger.info(f"Memories used last time: {used_memories_last_time}")

            # Filter out any overlap
            new_memories = [
                m for m in recent_memories
                if m not in used_memories_last_time
            ]

            # If all are duplicates, we skip tweeting
            if not new_memories:
                logger.info("All recent memories overlap with last time.")
                return ActivityResult(success=True, data={"message": "No new memories to tweet."})

            # 5) Build prompt referencing personality + objectives + the final set of memories
            prompt_text = self._build_chat_prompt(
                personality=personality_data,
                objectives=objectives_data,
                new_memories=new_memories
            )

            # 6) Use chat skill to generate the tweet text
            chat_response = await chat_skill.get_chat_completion(
                prompt=prompt_text,
                system_prompt=(
                    "You are an AI that composes tweets with the given personality and objectives. "
                    "Tweet must be under 280 chars."
                ),
                max_tokens=200
            )
            if not chat_response["success"]:
                return ActivityResult(success=False, error=chat_response["error"])

            tweet_text = chat_response["data"]["content"].strip()
            if len(tweet_text) > self.max_length:
                tweet_text = tweet_text[: self.max_length - 3] + "..."

            # 7) Post to Twitter via Composio
            post_result = self._post_tweet_via_composio(tweet_text)
            if not post_result["success"]:
                error_msg = post_result.get("error", "Unknown error posting tweet via Composio")
                logger.error(f"Tweet posting failed: {error_msg}")
                return ActivityResult(success=False, error=error_msg)

            tweet_id = post_result.get("tweet_id")
            tweet_link = f"https://twitter.com/{self.twitter_username}/status/{tweet_id}" if tweet_id else None

            # 8) Return success, storing the new memories in "data" so we can skip them next time
            logger.info(f"Successfully posted tweet about recent memories: {tweet_text[:50]}...")
            return ActivityResult(
                success=True,
                data={
                    "tweet_id": tweet_id,
                    "content": tweet_text,
                    "recent_memories_used": new_memories  # store these for next run
                },
                metadata={
                    "length": len(tweet_text),
                    "tweet_link": tweet_link,
                    "prompt_used": prompt_text,
                    "model": chat_response["data"].get("model"),
                    "finish_reason": chat_response["data"].get("finish_reason")
                }
            )

        except Exception as e:
            logger.error(f"Failed to post recent memories tweet: {e}", exc_info=True)
            return ActivityResult(success=False, error=str(e))

    def _get_memories_used_last_time(self, shared_data) -> List[str]:
        """
        Look in memory for the most recent successful run of this same activity.
        Return the list of 'recent_memories_used' from that run, or [] if none.
        """
        system_data = shared_data.get_category_data("system")
        memory_obj: Memory = system_data.get("memory_ref")
        if not memory_obj:
            from framework.main import DigitalBeing
            being = DigitalBeing()
            being.initialize()
            memory_obj = being.memory

        # Search in the last ~10 runs for this activity
        recent_activities = memory_obj.get_recent_activities(limit=10, offset=0)
        for act in recent_activities:
            if (act.get("activity_type") == "PostRecentMemoriesTweetActivity"
                    and act.get("success")):
                used = act.get("data", {}).get("recent_memories_used", [])
                if used:
                    return used
        return []

    def _get_character_config(self, shared_data) -> Dict[str, Any]:
        """
        Retrieve character_config from SharedData['system'] or re-init the Being if not found.
        """
        system_data = shared_data.get_category_data("system")
        maybe_config = system_data.get("character_config")
        if maybe_config:
            return maybe_config

        # fallback
        from framework.main import DigitalBeing
        being = DigitalBeing()
        being.initialize()
        return being.configs.get("character_config", {})

    def _get_recent_memories(self, shared_data, limit: int = 10) -> List[str]:
        """
        Pull up to 'limit' recent memory items (activities),
        ignoring certain activity types in self.ignored_activity_types.
        We'll just gather a short summary for each activity.
        """
        system_data = shared_data.get_category_data("system")
        memory_obj: Memory = system_data.get("memory_ref")

        if not memory_obj:
            from framework.main import DigitalBeing
            being = DigitalBeing()
            being.initialize()
            memory_obj = being.memory

        recent_activities = memory_obj.get_recent_activities(limit=50, offset=0)
        memories = []
        for act in recent_activities:
            act_type = act.get("activity_type")
            if act_type in self.ignored_activity_types:
                continue  # skip

            # Some minimal representation
            summary = f"{act_type} => {act.get('data', {})}"
            memories.append(summary)

            if len(memories) >= limit:
                break

        return memories

    def _build_chat_prompt(
        self,
        personality: Dict[str, Any],
        objectives: Dict[str, Any],
        new_memories: List[str]
    ) -> str:
        """
        Construct the user prompt: combine personality + objectives + the new memory summaries,
        and instruct the model to craft a short tweet.
        """
        # Personality lines
        trait_lines = [f"{t}: {v}" for t, v in personality.items()]
        personality_str = "\n".join(trait_lines)

        # Objectives lines
        objective_lines = []
        for k, v in objectives.items():
            objective_lines.append(f"{k}: {v}")
        objectives_str = "\n".join(objective_lines) if objective_lines else "(No objectives specified)"

        # Memories
        if new_memories:
            memories_str = "\n".join(f"- {txt}" for txt in new_memories)
        else:
            memories_str = "(No new memories)"

        prompt = (
            f"Our digital being has these personality traits:\n"
            f"{personality_str}\n\n"
            f"It also has these objectives:\n"
            f"{objectives_str}\n\n"
            f"Here are some new memories:\n"
            f"{memories_str}\n\n"
            f"Please craft a short tweet (under 280 chars) that references these memories, "
            f"reflects the personality and objectives, and ensures it's not repetitive or dull. "
            f"Keep it interesting, cohesive, and mindful of the overall tone.\n"
        )
        return prompt

    def _post_tweet_via_composio(self, tweet_text: str) -> Dict[str, Any]:
        """
        Same as your original PostTweetActivity approach to tweeting via Composio.
        """
        try:
            from framework.composio_integration import composio_manager
            logger.info(f"Posting tweet via Composio action='{self.composio_action}', text='{tweet_text[:50]}...'")

            response = composio_manager._toolset.execute_action(
                action=self.composio_action,
                params={"text": tweet_text},
                entity_id="MyDigitalBeing"
            )

            success_val = response.get("success", response.get("successfull"))
            if success_val:
                data_section = response.get("data", {})
                nested_data = data_section.get("data", {})
                tweet_id = nested_data.get("id")
                return {"success": True, "tweet_id": tweet_id}
            else:
                return {
                    "success": False,
                    "error": response.get("error", "Unknown or missing success key")
                }

        except Exception as e:
            logger.error(f"Error in Composio tweet post: {e}", exc_info=True)
            return {"success": False, "error": str(e)}


##### my_digital_being/activities/activity_suggest_new_activities.py #####
# activities/activity_suggest_new_activities.py

import logging
from typing import Any, Dict
from framework.activity_decorator import activity, ActivityBase, ActivityResult
from skills.skill_chat import chat_skill

# We import these so we can list out both manual + dynamic skill records
from framework.skill_config import DynamicComposioSkills
from framework.main import DigitalBeing

logger = logging.getLogger(__name__)

@activity(
    name="SuggestNewActivities",
    energy_cost=0.4,
    cooldown=259200,  # 3 days
    required_skills=["openai_chat"]
)
class SuggestNewActivities(ActivityBase):
    """
    Activity that examines the being's current objectives and constraints,
    then asks the LLM to propose new or modified Activities which may leverage
    any known skills (both manual-coded + dynamic from Composio).
    """

    def __init__(self):
        super().__init__()
        self.system_prompt = """You are an AI that helps brainstorm new or improved
Activities (Python-coded tasks) to achieve the being's goals, leveraging the skills
the system has available. The user will evaluate or build these later. Provide short,
actionable suggestions focusing on feasibility, alignment with constraints, and creativity.
If relevant, mention which skill(s) would be used for each suggestion.
        Do not plan on using API calls or making up URLs and rely on available skills for interacting with anything external to yourself."""

    async def execute(self, shared_data) -> ActivityResult:
        try:
            logger.info("Starting new activity suggestion process...")

            # 1) Initialize the chat skill
            if not await chat_skill.initialize():
                return ActivityResult(
                    success=False,
                    error="Failed to initialize openai_chat skill"
                )

            # 2) Gather the being + config
            being = DigitalBeing()
            being.initialize()
            char_cfg = being.configs.get("character_config", {})
            objectives = char_cfg.get("objectives", {})
            primary_obj = objectives.get("primary", "No primary objective found.")
            constraints_cfg = being.configs.get("activity_constraints", {})
            global_cons = constraints_cfg.get("global_constraints", "None specified")

            # 3) Gather all known skills (manual + dynamic)
            skills_config = being.configs.get("skills_config", {})

            # A. Manual-coded skills from skills_config.json
            manual_skill_list = []
            for skill_name, skill_info in skills_config.items():
                # skill_info can be dict or something else
                if isinstance(skill_info, dict):
                    # We'll build a short desc
                    desc = f"Skill: {skill_name}, enabled={skill_info.get('enabled')}"
                    # Add required keys, etc. if relevant
                    req_keys = skill_info.get('required_api_keys', [])
                    desc += f", required_api_keys={req_keys}"
                    meta = skill_info.get('metadata', {})
                    if meta:
                        desc += f", metadata={meta}"
                    manual_skill_list.append(desc)
                else:
                    # e.g. skip "default_llm_skill": "openai_chat"
                    pass

            # B. Dynamic (Composio) discovered skills
            dynamic_skills = DynamicComposioSkills.get_all_dynamic_skills()
            dynamic_skill_list = []
            for ds in dynamic_skills:
                d_name = ds["skill_name"]
                d_enabled = ds.get("enabled", True)
                d_req = ds.get("required_api_keys", [])
                d_meta = ds.get("metadata", {})
                desc = f"DynamicSkill: {d_name}, enabled={d_enabled}, required_api_keys={d_req}, metadata={d_meta}"
                dynamic_skill_list.append(desc)

            # 4) Combine skill descriptions into one block
            all_skills_block = "\n".join(manual_skill_list + dynamic_skill_list)
            if not all_skills_block.strip():
                all_skills_block = "(No known skills found)"

            # 5) Build final prompt
            prompt_text = (
                f"My primary objective: {primary_obj}\n"
                f"Global constraints or notes: {global_cons}\n\n"
                f"Known Skills:\n{all_skills_block}\n\n"
                f"Propose up to 3 new or modified Activities to help achieve my goal. "
                f"Highlight how each might use one or more of these skills (if relevant). "
                f"Keep suggestions short."
            )

            # 6) LLM call
            response = await chat_skill.get_chat_completion(
                prompt=prompt_text,
                system_prompt=self.system_prompt,
                max_tokens=300
            )
            if not response["success"]:
                return ActivityResult(success=False, error=response["error"])

            suggestions = response["data"]["content"]

            return ActivityResult(
                success=True,
                data={"suggestions": suggestions},
                metadata={
                    "model": response["data"]["model"],
                    "finish_reason": response["data"]["finish_reason"]
                }
            )

        except Exception as e:
            logger.error(f"Error in SuggestNewActivities: {e}")
            return ActivityResult(success=False, error=str(e))


##### my_digital_being/activities/activity_test.py #####
import logging
from typing import Dict, Any
from framework.activity_decorator import activity, ActivityBase, ActivityResult

logger = logging.getLogger(__name__)

@activity(
    name="Test",
    energy_cost=0.2,
    cooldown=300,
    required_skills=[]
)
class TestActivity(ActivityBase):
    def __init__(self):
        super().__init__()

    async def execute(self, shared_data) -> ActivityResult:
        try:
            # Example: This is the main logic for "Test" activity
            logger.info("Executing Test activity")
            # TODO: Actual logic goes here
            return ActivityResult(success=True, data={"message":"Test done"})
        except Exception as e:
            logger.error(f"Error in Test activity: {e}")
            return ActivityResult(success=False, error=str(e))

##### my_digital_being/server.py #####
"""
Digital Being Server implementation.
Implements:
 - /oauth_callback endpoint to finalize OAuth after user is redirected back
 - WebSocket commands for front-end
 - Pause/Resume logic
 - Checking is_configured for front-end
 - [ADDED] Returning 'enabled' status for each loaded activity
"""

import asyncio
import json
import logging
import os
import http
import mimetypes
from pathlib import Path
from typing import Dict, Any, Set, Optional, Union, Tuple
from datetime import datetime

import websockets
from websockets.server import serve
from websockets.legacy.server import WebSocketServerProtocol

from framework.main import DigitalBeing
from framework.api_management import api_manager
from framework.skill_config import DynamicComposioSkills

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class DigitalBeingServer:
    """Server for the Digital Being application."""

    def __init__(self, host: str = "0.0.0.0", port: int = 8000):
        self.host = host
        self.port = port
        self.clients: Set[WebSocketServerProtocol] = set()
        self.being = DigitalBeing()
        self.being_state: Dict[str, Any] = {}
        self.static_path = Path(__file__).parent / "static"

        # [ADDED] Additional flags for running/paused
        self.running = False
        self.paused = False

    async def initialize(self):
        """Initialize the digital being and start periodic updates."""
        logger.info("Initializing Digital Being...")
        self.being.initialize()  # load config, etc.

        self.running = True  # default "running"
        asyncio.create_task(self._periodic_state_update())
        asyncio.create_task(self._run_being_loop())

    async def _run_being_loop(self):
        """Main loop that calls the being's activities if running & not paused."""
        while True:
            try:
                if not self.running:
                    await asyncio.sleep(2)
                    continue

                if self.paused:
                    await asyncio.sleep(2)
                    continue

                if not self.being.is_configured():
                    # If not configured, do nothing in the main loop
                    await asyncio.sleep(2)
                    continue

                # Single-step approach for selecting an activity
                current_activity = self.being.activity_selector.select_next_activity()
                if current_activity:
                    logger.info(f"Executing activity: {current_activity.__class__.__name__}")
                    result = await self.being.execute_activity(current_activity)
                    if result and result.success:
                        self.being_state["last_activity"] = {
                            "name": current_activity.__class__.__name__,
                            "timestamp": datetime.now().isoformat(),
                            "success": True,
                        }
                    else:
                        self.being_state["last_activity"] = {
                            "name": current_activity.__class__.__name__,
                            "timestamp": datetime.now().isoformat(),
                            "success": False,
                            "error": (result.error if result else "Unknown error"),
                        }
                    await self.broadcast_state()

                self.being.state.update()
                self.being.memory.persist()
                await asyncio.sleep(5)

            except Exception as e:
                logger.error(f"Error in being loop: {e}")
                await asyncio.sleep(10)

    async def _periodic_state_update(self):
        """Periodically update and broadcast the being's state every second."""
        while True:
            try:
                current_state = self.being.state.get_current_state()
                # Provide 'configured' and 'paused' status
                current_state["configured"] = self.being.is_configured()
                current_state["paused"] = self.paused

                if current_state != self.being_state:
                    self.being_state = current_state
                    await self.broadcast_state()
                await asyncio.sleep(1)
            except Exception as e:
                logger.error(f"Error in periodic state update: {e}")
                await asyncio.sleep(5)

    async def register(self, websocket: WebSocketServerProtocol):
        self.clients.add(websocket)
        logger.info(f"Client connected. Total clients: {len(self.clients)}")

        # Send the current state right away
        await websocket.send(json.dumps({
            "type": "state_update",
            "data": self.being_state
        }))

    async def unregister(self, websocket: WebSocketServerProtocol):
        self.clients.discard(websocket)
        logger.info(f"Client disconnected. Total clients: {len(self.clients)}")

    async def serve_static_file(self, path: str, request_headers: Dict) -> Union[Tuple[http.HTTPStatus, list, bytes], None]:
        try:
            if path == "/ws":
                return None

            if path.startswith("/oauth_callback"):
                return await self.handle_oauth_http_callback(path)

            if not isinstance(path, str):
                return None

            request_path = "/" + path.lstrip("/")
            if request_path == "/":
                request_path = "/index.html"

            if request_path == "/ws":
                if (request_headers.get("Upgrade", "").lower() == "websocket"
                    and request_headers.get("Connection", "").lower() == "upgrade"):
                    logger.info("Valid WebSocket upgrade request")
                    return None
                logger.warning("Invalid WebSocket request")
                return http.HTTPStatus.BAD_REQUEST, [("Content-Type", "text/plain")], b"Invalid WebSocket request"

            file_path = self.static_path / request_path.lstrip("/")
            if not file_path.exists() or not file_path.is_file():
                logger.warning(f"File not found: {file_path}")
                return http.HTTPStatus.NOT_FOUND, [("Content-Type", "text/plain")], b"404 Not Found"

            content_type, _ = mimetypes.guess_type(str(file_path))
            if not content_type:
                content_type = "application/octet-stream"

            content = file_path.read_bytes()
            return http.HTTPStatus.OK, [
                ("Content-Type", content_type),
                ("Cache-Control", "public, max-age=3600"),
            ], content

        except Exception as e:
            logger.error(f"Error serving {path}: {e}")
            return http.HTTPStatus.INTERNAL_SERVER_ERROR, [
                ("Content-Type", "text/plain")
            ], b"Internal Server Error"

    async def handle_oauth_http_callback(self, path: str):
        """
        Handle GET /oauth_callback?status=success&connectedAccountId=...&appName=...
        Then finalize or store connection info, auto-fetch & register app actions, 
        and finally redirect to "/".
        """
        from urllib.parse import urlparse, parse_qs

        parsed = urlparse(path)
        query = parse_qs(parsed.query)

        status = query.get("status", [""])[0]
        connected_account_id = query.get("connectedAccountId", [None])[0]
        app_name = query.get("appName", [""])[0]
        code = query.get("code", [None])[0]

        if not connected_account_id:
            logger.error("Missing connectedAccountId param in /oauth_callback")
            body = b"Missing connectedAccountId param"
            return (
                http.HTTPStatus.BAD_REQUEST,
                [("Content-Type", "text/plain")],
                body,
            )

        logger.info(
            f"OAuth callback success for app={app_name}, connectedAccountId={connected_account_id}, status={status}"
        )

        try:
            if code:
                finalize_result = await api_manager.composio_manager.handle_oauth_callback(connected_account_id, code)
                logger.info(f"handle_oauth_callback returned: {finalize_result}")
            else:
                if app_name:
                    api_manager.composio_manager.mark_app_connected_without_code(app_name, connected_account_id)

            if app_name:
                logger.info(f"Auto-fetching actions for newly connected app: {app_name}")
                actions_result = await api_manager.list_actions_for_app(app_name)
                if actions_result.get("success"):
                    actions = actions_result.get("actions", [])
                    if actions:
                        logger.info(f"Discovered {len(actions)} actions for {app_name}, registering now...")
                        DynamicComposioSkills.register_composio_actions(app_name, actions)
                    else:
                        logger.warning(f"No actions found for {app_name}.")
                else:
                    logger.warning(f"Failed to fetch actions for {app_name}: {actions_result.get('error')}")

        except Exception as e:
            logger.error(f"Error finalizing/fetching actions for {app_name}: {e}", exc_info=True)

        redirect_body = b'<html><head><meta http-equiv="refresh" content="0;URL=\'/\'" /></head><body>Redirecting...</body></html>'
        return (
            http.HTTPStatus.OK,
            [("Content-Type", "text/html")],
            redirect_body,
        )

    async def handle_websocket(self, websocket: WebSocketServerProtocol, path: str):
        """Handle WebSocket connections at /ws."""
        try:
            if path != "/ws":
                logger.warning(f"Invalid WebSocket path: {path}")
                await websocket.close(code=1008, reason="Invalid path")
                return

            await self.register(websocket)
            try:
                async for message in websocket:
                    try:
                        data = json.loads(message)
                        await self.process_message(websocket, data)
                    except json.JSONDecodeError:
                        logger.error(f"Invalid JSON: {message}")
                    except Exception as e:
                        logger.error(f"Error processing WS message: {e}")
            except websockets.ConnectionClosed:
                logger.info("WebSocket closed normally")
            except Exception as e:
                logger.error(f"WebSocket error: {e}")
        finally:
            await self.unregister(websocket)

    async def process_message(self, websocket: WebSocketServerProtocol, data: Dict[str, Any]):
        try:
            message_type = data.get("type")
            if not message_type:
                logger.warning("No message type in WS data!")
                return

            if message_type == "get_state":
                await websocket.send(json.dumps({
                    "type": "state_update",
                    "data": self.being_state
                }))
            elif message_type == "command":
                command = data.get("command")
                if command:
                    resp = await self.handle_command(command, data.get("params", {}))
                    await websocket.send(json.dumps({
                        "type": "command_response",
                        "command": command,
                        "response": resp
                    }))
        except Exception as e:
            logger.error(f"Error in process_message: {e}")
            await websocket.send(json.dumps({
                "type": "error",
                "message": str(e)
            }))

    async def handle_command(self, command: str, params: Dict[str, Any]) -> Dict[str, Any]:
        logger.debug(f"handle_command: {command}, params={params}")
        try:
            if command == "pause":
                self.paused = True
                return {"success": True, "message": "Digital Being is paused."}
            elif command == "resume":
                self.paused = False
                return {"success": True, "message": "Digital Being resumed."}
            elif command == "stop_loop":
                self.running = False
                return {"success": True, "message": "Core loop stopped."}
            elif command == "start_loop":
                self.running = True
                return {"success": True, "message": "Core loop started."}

            elif command == "initiate_oauth":
                app_name = params.get("app_name")
                base_url = params.get("base_url", "http://localhost:8000")
                if not app_name:
                    return {"success": False, "error": "Missing app_name"}
                redirect_url = f"{base_url}/oauth_callback"
                try:
                    result = await api_manager.composio_manager.initiate_oauth_flow(app_name, redirect_url)
                    return result
                except Exception as e:
                    logger.error(f"init_oauth error: {e}")
                    return {"success": False, "error": str(e)}

            elif command == "get_composio_integrations":
                try:
                    integrations = await api_manager.composio_manager.list_available_integrations()
                    return {"success": True, "composio_integrations": integrations}
                except Exception as e:
                    logger.error(f"Error get_composio_integrations: {e}")
                    return {"success": False, "error": str(e), "composio_integrations": []}

            elif command == "get_api_key_status":
                skills_status = await api_manager.get_skill_status()
                return {"success": True, "skills": skills_status}

            elif command == "configure_api_key":
                skill_name = params.get("skill_name")
                key_name = params.get("key_name")
                api_key_value = params.get("api_key")
                if not all([skill_name, key_name, api_key_value]):
                    return {"success": False, "message": "Missing required params"}
                try:
                    result = await api_manager.set_api_key(skill_name, key_name, api_key_value)
                    return result
                except Exception as e:
                    return {"success": False, "message": str(e)}

            elif command == "get_system_status":
                memory_stats = {
                    "short_term_count": len(self.being.memory.short_term_memory),
                    "long_term_count": sum(len(x) for x in self.being.memory.long_term_memory.values()),
                    "total_activities": self.being.memory.get_activity_count(),
                }
                current_state = self.being.state.get_current_state()
                is_config = self.being.is_configured()

                return {
                    "success": True,
                    "memory": memory_stats,
                    "state": current_state,
                    "is_configured": is_config,
                    "config": self.being.configs
                }

            # [ADDED] 'get_activities' now returns 'enabled' from activity_constraints["activities_config"]
            elif command == "get_activities":
                acts = self.being.activity_loader.get_all_activities()
                info = {}
                # This dictionary might be in self.being.configs["activity_constraints"]["activities_config"]
                activities_config = {}
                if "activity_constraints" in self.being.configs and \
                   "activities_config" in self.being.configs["activity_constraints"]:
                    activities_config = self.being.configs["activity_constraints"]["activities_config"]

                for module_name, cls in acts.items():
                    class_name = cls.__name__
                    # If user has an "enabled" setting, read it. Otherwise default to True
                    is_enabled = True
                    if class_name in activities_config:
                        is_enabled = bool(activities_config[class_name].get("enabled", True))

                    info[module_name] = {
                        "name": class_name,
                        "energy_cost": cls.energy_cost,
                        "cooldown": cls.cooldown,
                        "required_skills": cls.required_skills,
                        "last_execution": (cls.last_execution.isoformat() if cls.last_execution else None),
                        "enabled": is_enabled
                    }
                return {"success": True, "activities": info}

            elif command == "get_config":
                return {"success": True, "config": self.being.configs}

            elif command == "update_config":
                section = params.get("section")
                key = params.get("key")
                value = params.get("value")

                if not section or not key:
                    return {"success": False, "message": "Both 'section' and 'key' are required."}

                # Map sections to their respective config files
                section_to_file = {
                    "character_config": "character_config.json",
                    "skills_config": "skills_config.json",
                    "activity_constraints": "activity_constraints.json"
                }

                config_file_name = section_to_file.get(section)
                if not config_file_name:
                    return {"success": False, "message": f"Unknown configuration section: {section}"}

                config_path = Path(self.being.config_path) / config_file_name

                # Load existing config
                try:
                    if config_path.exists():
                        with open(config_path, 'r') as f:
                            current_config = json.load(f)
                    else:
                        current_config = {}
                except json.JSONDecodeError as je:
                    logger.error(f"Failed to parse {config_file_name}: {je}")
                    return {"success": False, "message": f"Invalid JSON format in {config_file_name}."}
                except Exception as e:
                    logger.error(f"Error loading {config_file_name}: {e}")
                    return {"success": False, "message": f"Error loading {config_file_name}."}

                # Update the config
                current_config[key] = value

                # Write back to the config file
                try:
                    with open(config_path, 'w') as f:
                        json.dump(current_config, f, indent=2)
                except Exception as e:
                    logger.error(f"Failed to write to {config_file_name}: {e}")
                    return {"success": False, "message": f"Failed to write to {config_file_name}."}

                # Update in-memory config
                self.being.configs[section][key] = value

                logger.info(f"Updated config: [{section}] {key} = {value}")

                return {"success": True, "message": f"Configuration '{key}' updated successfully."}

            elif command == "get_activity_history":
                limit = params.get("limit", 10)
                offset = params.get("offset", 0)
                recents = self.being.memory.get_recent_activities(limit=limit, offset=offset)
                total = self.being.memory.get_activity_count()
                return {
                    "success": True,
                    "activities": recents,
                    "has_more": total > (offset + limit),
                    "total": total,
                }

            elif command == "get_composio_app_actions":
                app_name = params.get("app_name")
                result = await api_manager.list_actions_for_app(app_name)
                if result.get("success"):
                    DynamicComposioSkills.register_composio_actions(app_name, result.get("actions", []))
                return result

            elif command == "get_all_skills":
                config_skills = self.being.configs.get("skills_config", {})
                manual_skills_list = []
                for skill_name, skill_info in config_skills.items():
                    if not isinstance(skill_info, dict):
                        logger.debug(f"Skipping non-dict skill config: {skill_name} => {skill_info}")
                        continue
                    manual_skills_list.append({
                        "skill_name": skill_name,
                        "enabled": bool(skill_info.get("enabled", False)),
                        "metadata": skill_info
                    })

                dynamic_skills = DynamicComposioSkills.get_all_dynamic_skills()
                all_skills = manual_skills_list + dynamic_skills
                return {"success": True, "skills": all_skills}

            elif command == "get_activity_code":
                from framework.activity_loader import read_activity_code
                activity_name = params.get("activity_name")
                code_str = read_activity_code(activity_name)
                if code_str is None:
                    return {"success": False, "message": f"Could not read code for {activity_name}"}
                return {"success": True, "code": code_str}

            elif command == "save_activity_code":
                from framework.activity_loader import write_activity_code
                activity_name = params.get("activity_name")
                new_code = params.get("new_code")
                ok = write_activity_code(activity_name, new_code)
                if not ok:
                    return {"success": False, "message": "Failed to save code"}
                self.being.activity_loader.reload_activities()
                return {"success": True, "message": "Code updated and reloaded"}

            elif command == "save_onboarding_data":
                """
                Expects 'character', 'skills', and 'constraints' from front-end. 
                If you also want to let user enable/disable activities in front-end, 
                pass 'constraints.activities_config' with the new enabled settings here.
                """
                try:
                    char_data = params.get("character", {})
                    skills_data = params.get("skills", {})
                    constraints_data = params.get("constraints", {})

                    char_path = Path(self.being.config_path) / "character_config.json"
                    skill_path = Path(self.being.config_path) / "skills_config.json"
                    actc_path = Path(self.being.config_path) / "activity_constraints.json"

                    existing_char = {}
                    existing_skills = {}
                    existing_actc = {}

                    if char_path.exists():
                        existing_char = json.loads(char_path.read_text(encoding="utf-8"))
                    if skill_path.exists():
                        existing_skills = json.loads(skill_path.read_text(encoding="utf-8"))
                    if actc_path.exists():
                        existing_actc = json.loads(actc_path.read_text(encoding="utf-8"))

                    existing_char.update(char_data)
                    existing_skills.update(skills_data)
                    # Merge constraints
                    for k, v in constraints_data.items():
                        existing_actc[k] = v

                    # Possibly set "setup_complete": true
                    existing_char["setup_complete"] = True

                    char_path.write_text(json.dumps(existing_char, indent=2), encoding="utf-8")
                    skill_path.write_text(json.dumps(existing_skills, indent=2), encoding="utf-8")
                    actc_path.write_text(json.dumps(existing_actc, indent=2), encoding="utf-8")

                    # Reload in memory
                    self.being.configs["character_config"] = existing_char
                    self.being.configs["skills_config"] = existing_skills
                    self.being.configs["activity_constraints"] = existing_actc

                    return {"success": True, "message": "Onboarding data saved."}
                except Exception as e:
                    logger.error(f"Error saving onboarding data: {e}", exc_info=True)
                    return {"success": False, "message": str(e)}

        except Exception as e:
            logger.error(f"handle_command {command} error: {e}")
            return {"success": False, "message": str(e)}

        return {"success": False, "message": "Unknown command"}


    async def broadcast_state(self):
        """Broadcast the current being_state to all connected WebSocket clients."""
        if not self.clients:
            return
        message = json.dumps({
            "type": "state_update",
            "data": self.being_state
        })
        disconnected_clients = set()
        for client in self.clients:
            try:
                await client.send(message)
            except websockets.ConnectionClosed:
                logger.info("Client disconnected during broadcast")
                disconnected_clients.add(client)
            except Exception as e:
                logger.error(f"Error broadcasting to client: {e}")
                disconnected_clients.add(client)

        for dc in disconnected_clients:
            await self.unregister(dc)

    async def start_server(self):
        """Start the server using websockets.serve()."""
        try:
            await self.initialize()
            async with serve(
                self.handle_websocket,
                self.host,
                self.port,
                process_request=self.serve_static_file
            ):
                logger.info(f"Server started on ws://{self.host}:{self.port}")
                await asyncio.Future()  # run forever
        except Exception as e:
            logger.error(f"Failed to start server: {e}")
            raise

if __name__ == "__main__":
    server = DigitalBeingServer()
    asyncio.run(server.start_server())