"""Store class for managing users.""" import asyncio import uuid from typing import Optional from uuid import UUID from server.auth.token_manager import TokenManager from server.constants import ( DEFAULT_V1_ENABLED, LITE_LLM_API_URL, ORG_SETTINGS_VERSION, PERSONAL_WORKSPACE_VERSION_TO_MODEL, get_default_llm_base_url, get_default_llm_model, ) from server.logger import logger from sqlalchemy import select, text from sqlalchemy.orm import selectinload from storage.database import a_session_maker from storage.encrypt_utils import ( decrypt_legacy_model, decrypt_legacy_value, encrypt_legacy_value, ) from storage.org import Org from storage.org_member import OrgMember from storage.role import Role from storage.role_store import RoleStore from storage.user import User from storage.user_settings import UserSettings from utils.identity import resolve_display_name from openhands.sdk.settings import AGENT_SETTINGS_SCHEMA_VERSION # The max possible time to wait for another process to finish creating a user before retrying _REDIS_CREATE_TIMEOUT_SECONDS = 30 # The delay to wait for another process to finish creating a user before trying to load again _RETRY_LOAD_DELAY_SECONDS = 2 # Redis key prefix for user creation locks _REDIS_USER_CREATION_KEY_PREFIX = 'create_user:' class UserStore: """Store for managing users.""" @staticmethod async def create_user( user_id: str, user_info: dict, role_id: Optional[int] = None, ) -> User | None: """Create a new user.""" async with a_session_maker() as session: # create personal org org = Org( id=uuid.UUID(user_id), name=f'user_{user_id}_org', contact_name=resolve_display_name(user_info) or user_info.get('preferred_username', ''), contact_email=user_info['email'], v1_enabled=True, ) session.add(org) settings = await UserStore.create_default_settings( org_id=str(org.id), user_id=user_id ) if not settings: return None from storage.org_store import OrgStore org_kwargs = OrgStore.get_kwargs_from_settings(settings) for key, value in org_kwargs.items(): if hasattr(org, key): setattr(org, key, value) user_kwargs = UserStore.get_kwargs_from_settings(settings) user = User( id=uuid.UUID(user_id), current_org_id=org.id, role_id=role_id, **user_kwargs, ) user.email = user_info.get('email') user.email_verified = user_info.get('email_verified') session.add(user) role = await RoleStore.get_role_by_name('owner') if role is None: raise ValueError('Owner role not found in database') from storage.org_member_store import OrgMemberStore org_member_kwargs = OrgMemberStore.get_kwargs_from_settings(settings) org_member = OrgMember( org_id=org.id, user_id=user.id, role_id=role.id, # owner of your own org. status='active', **org_member_kwargs, ) session.add(org_member) await session.commit() await session.refresh(user) await session.refresh(user, ['org_members']) # load org_members return user @staticmethod def _get_redis_client(): """Get the shared async Redis client from enterprise storage.""" from storage.redis import get_redis_client_async return get_redis_client_async() @staticmethod async def _acquire_user_creation_lock(user_id: str) -> bool: """Attempt to acquire a distributed lock for user creation. Returns True if the lock was acquired or if Redis is unavailable (fallback to no locking). Returns False if another process holds the lock. """ from storage.redis import redis_exceptions redis_client = UserStore._get_redis_client() try: user_key = f'{_REDIS_USER_CREATION_KEY_PREFIX}{user_id}' lock_acquired = await redis_client.set( user_key, 1, nx=True, ex=_REDIS_CREATE_TIMEOUT_SECONDS ) return bool(lock_acquired) except redis_exceptions.RedisError: logger.warning( 'user_store:_acquire_user_creation_lock:redis_error', extra={'user_id': user_id}, ) return True # Proceed without locking on error @staticmethod async def _release_user_creation_lock(user_id: str) -> bool: """Release the distributed lock for user creation. Returns True if the lock was released or if Redis is unavailable. Returns False if the lock could not be released. """ from storage.redis import redis_exceptions redis_client = UserStore._get_redis_client() try: user_key = f'{_REDIS_USER_CREATION_KEY_PREFIX}{user_id}' deleted = await redis_client.delete(user_key) return bool(deleted) except redis_exceptions.RedisError: logger.warning( 'user_store:_release_user_creation_lock:redis_error', extra={'user_id': user_id}, ) return True # Proceed without locking on error @staticmethod async def migrate_user( user_id: str, user_settings: UserSettings, user_info: dict, ) -> User | None: kwargs = decrypt_legacy_model( [ 'llm_api_key', 'llm_api_key_for_byor', 'search_api_key', 'sandbox_api_key', ], user_settings, ) decrypted_user_settings = UserSettings(**kwargs) async with a_session_maker() as session: # Check if user has completed billing sessions to enable BYOR export from storage.billing_session import BillingSession result = await session.execute( select(BillingSession).filter( BillingSession.user_id == user_id, BillingSession.status == 'completed', ) ) has_completed_billing = result.scalars().first() is not None # create personal org org = Org( id=uuid.UUID(user_id), name=f'user_{user_id}_org', org_version=user_settings.user_version, contact_name=resolve_display_name(user_info) or user_info.get('username', ''), contact_email=user_info['email'], byor_export_enabled=has_completed_billing, ) session.add(org) from storage.lite_llm_manager import LiteLlmManager logger.debug( 'user_store:migrate_user:calling_litellm_migrate_entries', extra={'user_id': user_id}, ) await LiteLlmManager.migrate_entries( str(org.id), user_id, decrypted_user_settings, ) logger.debug( 'user_store:migrate_user:done_litellm_migrate_entries', extra={'user_id': user_id}, ) custom_settings = UserStore._has_custom_settings( decrypted_user_settings, user_settings.user_version ) # Migrate stripe customer (pass session to avoid FK violation) # avoids circular reference. This migrate method is temporary until all users are migrated. from integrations.stripe_service import migrate_customer logger.debug( 'user_store:migrate_user:calling_stripe_migrate_customer', extra={'user_id': user_id}, ) await migrate_customer(session, user_id, org) logger.debug( 'user_store:migrate_user:done_stripe_migrate_customer', extra={'user_id': user_id}, ) from storage.org_store import OrgStore org_kwargs = UserStore._get_org_kwargs_for_migration( decrypted_user_settings, custom_settings=custom_settings, ) for key, value in org_kwargs.items(): if hasattr(org, key): setattr(org, key, value) # Apply DEFAULT_V1_ENABLED for migrated orgs if v1_enabled was not set if org.v1_enabled is None: org.v1_enabled = DEFAULT_V1_ENABLED user_kwargs = UserStore.get_kwargs_from_user_settings( decrypted_user_settings ) user_kwargs.pop('id', None) user = User( id=uuid.UUID(user_id), current_org_id=org.id, role_id=None, **user_kwargs, ) session.add(user) logger.debug( 'user_store:migrate_user:calling_get_role_by_name', extra={'user_id': user_id}, ) role = await RoleStore.get_role_by_name('owner') logger.debug( 'user_store:migrate_user:done_get_role_by_name', extra={'user_id': user_id}, ) if role is None: raise ValueError('Owner role not found in database') from storage.org_member_store import OrgMemberStore org_member_kwargs = OrgMemberStore.get_kwargs_from_user_settings( decrypted_user_settings ) if not custom_settings: org_member_kwargs['agent_settings_diff'] = ( OrgStore.get_agent_settings_from_org(org).model_dump(mode='json') ) org_member = OrgMember( org_id=org.id, user_id=user.id, role_id=role.id, # owner of your own org. status='active', **org_member_kwargs, ) session.add(org_member) # Mark the old user_settings as migrated instead of deleting user_settings.already_migrated = True await session.merge(user_settings) await session.flush() logger.debug( 'user_store:migrate_user:session_flush_complete', extra={'user_id': user_id}, ) user_uuid = uuid.UUID(user_id) # need to migrate conversation metadata await session.execute( text(""" INSERT INTO conversation_metadata_saas (conversation_id, user_id, org_id) SELECT conversation_id, :user_uuid, :user_uuid FROM conversation_metadata WHERE user_id = :user_id_text """), {'user_uuid': user_uuid, 'user_id_text': user_id}, ) # Update stripe_customers await session.execute( text( 'UPDATE stripe_customers SET org_id = :org_id WHERE keycloak_user_id = :user_id' ), {'org_id': user_uuid, 'user_id': user_id}, ) # Update slack_users await session.execute( text( 'UPDATE slack_users SET org_id = :org_id WHERE keycloak_user_id = :user_id' ), {'org_id': user_uuid, 'user_id': user_id}, ) # Update slack_conversation await session.execute( text( 'UPDATE slack_conversation SET org_id = :org_id WHERE keycloak_user_id = :user_id' ), {'org_id': user_uuid, 'user_id': user_id}, ) # Update api_keys await session.execute( text('UPDATE api_keys SET org_id = :org_id WHERE user_id = :user_id'), {'org_id': user_uuid, 'user_id': user_id}, ) # Update custom_secrets await session.execute( text( 'UPDATE custom_secrets SET org_id = :org_id WHERE keycloak_user_id = :user_id' ), {'org_id': user_uuid, 'user_id': user_id}, ) # Update billing_sessions await session.execute( text( 'UPDATE billing_sessions SET org_id = :org_id WHERE user_id = :user_id' ), {'org_id': user_uuid, 'user_id': user_id}, ) await session.commit() await session.refresh(user) await session.refresh(user, ['org_members']) # load org_members logger.debug( 'user_store:migrate_user:session_committed', extra={'user_id': user_id}, ) return user @staticmethod async def downgrade_user(user_id: str) -> UserSettings | None: """This method can be removed once orgs is established - probably after Feb 15 2026 Downgrade a migrated user back to the pre-migration state. This reverses the migrate_user operation: 1. Get the user's settings from user_settings table (migrated users) or create new user_settings from org_members table (new sign-ups) 2. Call LiteLlmManager.downgrade_entries to revert LiteLLM state 3. Copy user_id from conversation_metadata_saas to conversation_metadata 4. Delete conversation_metadata_saas entries 5. Reset org_id columns in related tables (stripe_customers, slack_users, etc.) 6. Delete the org_member and org entries 7. Delete the user entry 8. Set already_migrated=False on user_settings For new sign-ups (users who registered after migration was deployed), there won't be an existing user_settings entry. In this case, we fall back to the org_members table to get the user's API keys and settings, and create a new user_settings entry for them. Args: user_id: The Keycloak user ID to downgrade Returns: The user_settings if downgrade was successful, None otherwise. Returns None if the org has multiple members (not a personal org). """ logger.info( 'user_store:downgrade_user:start', extra={'user_id': user_id}, ) async with a_session_maker() as session: # Get the user and their org_member result = await session.execute( select(User) .options(selectinload(User.org_members)) .filter(User.id == uuid.UUID(user_id)) ) user = result.scalars().first() if not user: logger.warning( 'user_store:downgrade_user:user_not_found', extra={'user_id': user_id}, ) return None # Get the user's personal org (org_id == user_id) result = await session.execute( select(Org).filter(Org.id == uuid.UUID(user_id)) ) org = result.scalars().first() if not org: logger.warning( 'user_store:downgrade_user:org_not_found', extra={'user_id': user_id}, ) return None # Get org_members for this org - should only be one for personal orgs result = await session.execute( select(OrgMember).filter(OrgMember.org_id == org.id) ) org_members = result.scalars().all() if len(org_members) != 1: logger.error( 'user_store:downgrade_user:unexpected_org_members_count', extra={ 'user_id': user_id, 'org_id': str(org.id), 'org_members_count': len(org_members), }, ) return None org_member = org_members[0] # Get the user_settings (for migrated users) result = await session.execute( select(UserSettings).filter( UserSettings.keycloak_user_id == user_id, UserSettings.already_migrated.is_(True), ) ) user_settings = result.scalars().first() # For new sign-ups after migration, user_settings won't exist # Fall back to getting data from org_members if user_settings: if org_member.llm_api_key and org_member.llm_api_key.get_secret_value(): user_settings.llm_api_key = encrypt_legacy_value( org_member.llm_api_key.get_secret_value() ) if ( org_member.llm_api_key_for_byor and org_member.llm_api_key_for_byor.get_secret_value() ): user_settings.llm_api_key_for_byor = encrypt_legacy_value( org_member.llm_api_key_for_byor.get_secret_value() ) logger.info( 'user_store:downgrade_user:updated_user_settings_from_org_member', extra={'user_id': user_id}, ) else: # Create a new user_settings entry from OrgMember, User, and Org data # This is needed for new sign-ups who don't have user_settings user_settings = UserStore._create_user_settings_from_entities( user_id, org_member, user, org ) session.add(user_settings) logger.info( 'user_store:downgrade_user:created_user_settings_from_org_member', extra={'user_id': user_id}, ) await session.flush() # Call LiteLLM downgrade from storage.lite_llm_manager import LiteLlmManager logger.debug( 'user_store:downgrade_user:calling_litellm_downgrade_entries', extra={'user_id': user_id}, ) encrypted_fields = [ 'llm_api_key', 'llm_api_key_for_byor', 'search_api_key', 'sandbox_api_key', ] for field in encrypted_fields: value = getattr(user_settings, field, None) if value: try: value = decrypt_legacy_value(value) setattr(user_settings, field, value) except Exception: pass await LiteLlmManager.downgrade_entries( str(org.id), user_id, user_settings, ) logger.debug( 'user_store:downgrade_user:done_litellm_downgrade_entries', extra={'user_id': user_id}, ) user_uuid = uuid.UUID(user_id) # Step 3: Copy user_id from conversation_metadata_saas to conversation_metadata # This ensures any conversations created after migration have their user_id # preserved in the original table before we delete the saas entries await session.execute( text(""" UPDATE conversation_metadata SET user_id = :user_id WHERE conversation_id IN ( SELECT conversation_id FROM conversation_metadata_saas WHERE user_id = :user_uuid ) """), {'user_id': user_id, 'user_uuid': user_uuid}, ) # Step 4: Delete conversation_metadata_saas entries await session.execute( text('DELETE FROM conversation_metadata_saas WHERE user_id = :user_id'), {'user_id': user_uuid}, ) # Step 5: Reset org_id columns in related tables # Reset stripe_customers await session.execute( text( 'UPDATE stripe_customers SET org_id = NULL WHERE org_id = :org_id' ), {'org_id': user_uuid}, ) # Reset slack_users await session.execute( text('UPDATE slack_users SET org_id = NULL WHERE org_id = :org_id'), {'org_id': user_uuid}, ) # Reset slack_conversation await session.execute( text( 'UPDATE slack_conversation SET org_id = NULL WHERE org_id = :org_id' ), {'org_id': user_uuid}, ) # Reset api_keys await session.execute( text('UPDATE api_keys SET org_id = NULL WHERE org_id = :org_id'), {'org_id': user_uuid}, ) # Reset custom_secrets await session.execute( text('UPDATE custom_secrets SET org_id = NULL WHERE org_id = :org_id'), {'org_id': user_uuid}, ) # Reset billing_sessions await session.execute( text( 'UPDATE billing_sessions SET org_id = NULL WHERE org_id = :org_id' ), {'org_id': user_uuid}, ) # Step 6: Delete org_member entries for this org await session.execute( text('DELETE FROM org_member WHERE org_id = :org_id'), {'org_id': user_uuid}, ) # Step 7: Delete the user entry await session.execute( text('DELETE FROM "user" WHERE id = :user_id'), {'user_id': user_uuid}, ) # Delete the org entry await session.execute( text('DELETE FROM org WHERE id = :org_id'), {'org_id': user_uuid}, ) # Step 8: Set already_migrated=False on user_settings and encrypt fields user_settings.already_migrated = False # Re-encrypt the sensitive fields before storing in the DB encrypt_keys = [ 'llm_api_key', 'llm_api_key_for_byor', 'search_api_key', 'sandbox_api_key', ] for key in encrypt_keys: value = getattr(user_settings, key, None) if value is not None and not _is_legacy_value_encrypted(value): setattr(user_settings, key, encrypt_legacy_value(value)) await session.merge(user_settings) await session.commit() logger.info( 'user_store:downgrade_user:complete', extra={'user_id': user_id}, ) return user_settings @staticmethod async def get_user_by_id(user_id: str) -> Optional[User]: """Get user by Keycloak user ID.""" async with a_session_maker() as session: result = await session.execute( select(User) .options(selectinload(User.org_members)) .filter(User.id == uuid.UUID(user_id)) ) user = result.scalars().first() if user: return user # Check if we need to migrate from user_settings while not await UserStore._acquire_user_creation_lock(user_id): # The user is already being created in another thread / process logger.info( 'user_store:create_default_settings:waiting_for_lock', extra={'user_id': user_id}, ) await asyncio.sleep(_RETRY_LOAD_DELAY_SECONDS) try: # Check for user again as migration could have happened while trying to get the lock. result = await session.execute( select(User) .options(selectinload(User.org_members)) .filter(User.id == uuid.UUID(user_id)) ) user = result.scalars().first() if user: return user result = await session.execute( select(UserSettings).filter( UserSettings.keycloak_user_id == user_id, UserSettings.already_migrated.is_(False), ) ) user_settings = result.scalars().first() if user_settings: token_manager = TokenManager() user_info = await token_manager.get_user_info_from_user_id(user_id) if not user_info: logger.warning( 'user_store:get_user_by_id:failed_to_get_user_info', extra={'user_id': user_id}, ) return None user = await UserStore.migrate_user( user_id, user_settings, user_info, ) return user else: return None finally: await UserStore._release_user_creation_lock(user_id) @staticmethod async def get_user_by_email(email: str) -> Optional[User]: """Get user by email address. This method looks up a user by their email address. Note that email addresses may not be unique across all users in rare cases. Args: email: The email address to search for Returns: User: The user with the matching email, or None if not found """ if not email: return None async with a_session_maker() as session: result = await session.execute( select(User) .options(selectinload(User.org_members)) .filter(User.email == email.lower().strip()) ) return result.scalars().first() @staticmethod async def list_users() -> list[User]: """List all users.""" async with a_session_maker() as session: result = await session.execute(select(User)) return list(result.scalars().all()) @staticmethod async def update_current_org(user_id: str, org_id: UUID) -> Optional[User]: """Update the user's current organization. Args: user_id: The user's ID (Keycloak user ID) org_id: The organization ID to set as current Returns: User: The updated user object, or None if user not found """ async with a_session_maker() as session: result = await session.execute( select(User).filter(User.id == uuid.UUID(user_id)).with_for_update() ) user = result.scalars().first() if not user: return None user.current_org_id = org_id await session.commit() await session.refresh(user) return user @staticmethod async def mark_onboarding_completed(user_id: str) -> Optional[User]: """Mark the user's onboarding as completed. Args: user_id: The user's ID (Keycloak user ID) Returns: User: The updated user object, or None if user not found """ async with a_session_maker() as session: result = await session.execute( select(User).filter(User.id == uuid.UUID(user_id)).with_for_update() ) user = result.scalars().first() if not user: logger.warning( 'mark_onboarding_completed:user_not_found', extra={'user_id': user_id}, ) return None user.onboarding_completed = True await session.commit() await session.refresh(user) logger.info( 'mark_onboarding_completed:success', extra={'user_id': user_id}, ) return user @staticmethod async def get_first_owner_in_org(org_id: UUID) -> Optional[User]: """Get the first owner in an organization who accepted the Terms of Service. This user is considered the super admin for that org in self-hosted deployments. The super admin is identified as the owner with the earliest accepted_tos timestamp. Args: org_id: The organization UUID Returns: User: The first owner to accept TOS in this org, or None if not found. """ async with a_session_maker() as session: result = await session.execute( select(User) .join(OrgMember, OrgMember.user_id == User.id) .join(Role, Role.id == OrgMember.role_id) .filter( OrgMember.org_id == org_id, Role.name == 'owner', User.accepted_tos.isnot(None), ) .order_by(User.accepted_tos.asc()) .limit(1) ) return result.scalars().first() @staticmethod async def backfill_contact_name(user_id: str, user_info: dict) -> None: """Update contact_name on the personal org if it still has a username-style value. Called during login to gradually fix existing users whose contact_name was stored as their username (before the resolve_display_name fix). Preserves custom values that were set via the PATCH endpoint. """ real_name = resolve_display_name(user_info) if not real_name: logger.debug( 'backfill_contact_name:no_real_name', extra={'user_id': user_id}, ) return preferred_username = user_info.get('preferred_username', '') username = user_info.get('username', '') async with a_session_maker() as session: result = await session.execute( select(Org).filter(Org.id == uuid.UUID(user_id)) ) org = result.scalars().first() if not org: logger.debug( 'backfill_contact_name:org_not_found', extra={'user_id': user_id}, ) return if org.contact_name in (preferred_username, username): logger.info( 'backfill_contact_name:updated', extra={ 'user_id': user_id, 'old': org.contact_name, 'new': real_name, }, ) org.contact_name = real_name await session.commit() @staticmethod async def update_user_email( user_id: str, email: str | None = None, email_verified: bool | None = None, ) -> None: """Unconditionally update User.email and/or email_verified. Unlike backfill_user_email(), this overwrites existing values. No-op when both arguments are None. Missing user is logged as a warning and ignored. """ if email is None and email_verified is None: return async with a_session_maker() as session: result = await session.execute( select(User).filter(User.id == uuid.UUID(user_id)) ) user = result.scalars().first() if not user: logger.warning( 'update_user_email:user_not_found', extra={'user_id': user_id}, ) return if email is not None: user.email = email if email_verified is not None: user.email_verified = email_verified logger.info( 'update_user_email:updated', extra={ 'user_id': user_id, 'email_set': email is not None, 'email_verified_set': email_verified is not None, }, ) await session.commit() @staticmethod async def backfill_user_email(user_id: str, user_info: dict) -> None: """Set User.email and email_verified from IDP if they are still NULL. Called during login to gradually fix existing users whose email was never persisted on the User record. Preserves non-NULL values (e.g. if a user manually changed their email). """ async with a_session_maker() as session: result = await session.execute( select(User).filter(User.id == uuid.UUID(user_id)) ) user = result.scalars().first() if not user: logger.debug( 'backfill_user_email:user_not_found', extra={'user_id': user_id}, ) return updated = False if user.email is None: user.email = user_info.get('email') updated = True if user.email_verified is None: user.email_verified = user_info.get('email_verified', False) updated = True if updated: logger.info( 'backfill_user_email:updated', extra={ 'user_id': user_id, 'email_set': user.email is not None, 'email_verified_set': user.email_verified is not None, }, ) await session.commit() # Prevent circular imports from typing import TYPE_CHECKING if TYPE_CHECKING: from openhands.app_server.settings.settings_models import Settings @staticmethod async def create_default_settings( org_id: str, user_id: str, create_user: bool = True ) -> Optional['Settings']: logger.info( 'UserStore:create_default_settings:start', extra={'org_id': org_id, 'user_id': user_id}, ) # You must log in before you get default settings if not org_id: return None from openhands.app_server.settings.settings_models import Settings default_settings = Settings( language='en', enable_proactive_conversation_starters=True ) default_settings.v1_enabled = DEFAULT_V1_ENABLED from storage.lite_llm_manager import LiteLlmManager settings = await LiteLlmManager.create_entries( org_id, user_id, default_settings, create_user ) if not settings: logger.info( 'UserStore:create_default_settings:litellm_create_failed', extra={'org_id': org_id}, ) return None return settings @staticmethod def get_kwargs_from_settings(settings: 'Settings'): kwargs = { normalized: getattr(settings, normalized) for c in User.__table__.columns if (normalized := c.name.lstrip('_')) and hasattr(settings, normalized) } return kwargs @staticmethod def get_kwargs_from_user_settings(user_settings: UserSettings): kwargs = { normalized: getattr(user_settings, normalized) for c in User.__table__.columns if (normalized := c.name.lstrip('_')) and hasattr(user_settings, normalized) } return kwargs @staticmethod def _create_user_settings_from_entities( user_id: str, org_member: OrgMember, user: User, org: Org ) -> UserSettings: """Create UserSettings from OrgMember, User, and Org data. Uses OrgMember values first. If an OrgMember field is None and there's a corresponding "default_" field in Org, use the Org value. Also pulls relevant fields from User. Args: user_id: The Keycloak user ID org_member: The OrgMember entity user: The User entity org: The Org entity Returns: A new UserSettings object populated from the entities """ from storage.org_store import OrgStore member_agent_settings_diff = dict(org_member.agent_settings_diff) org_agent_settings = OrgStore.get_agent_settings_from_org(org) agent_settings = { **org_agent_settings.model_dump(mode='json'), **member_agent_settings_diff, } member_conversation_settings_diff = dict(org_member.conversation_settings_diff) org_conversation_settings = OrgStore.get_conversation_settings_from_org(org) conversation_settings = { **org_conversation_settings.model_dump(mode='json'), **member_conversation_settings_diff, } return UserSettings( keycloak_user_id=user_id, llm_api_key=org_member.llm_api_key.get_secret_value() if org_member.llm_api_key else None, llm_api_key_for_byor=org_member.llm_api_key_for_byor.get_secret_value() if org_member.llm_api_key_for_byor else None, accepted_tos=user.accepted_tos, enable_sound_notifications=user.enable_sound_notifications, language=user.language, user_consents_to_analytics=user.user_consents_to_analytics, email=user.email, email_verified=user.email_verified, git_user_name=user.git_user_name, git_user_email=user.git_user_email, remote_runtime_resource_factor=org.remote_runtime_resource_factor, billing_margin=org.billing_margin, enable_proactive_conversation_starters=org.enable_proactive_conversation_starters, sandbox_base_container_image=org.sandbox_base_container_image, sandbox_runtime_container_image=org.sandbox_runtime_container_image, user_version=org.org_version, search_api_key=org.search_api_key.get_secret_value() if org.search_api_key else None, sandbox_api_key=org.sandbox_api_key.get_secret_value() if org.sandbox_api_key else None, max_budget_per_task=org.max_budget_per_task, v1_enabled=org.v1_enabled, sandbox_grouping_strategy=org.sandbox_grouping_strategy, agent_settings=agent_settings, conversation_settings=conversation_settings, already_migrated=False, ) @staticmethod def _get_org_kwargs_for_migration( user_settings: UserSettings, *, custom_settings: bool ) -> dict: from storage.org_store import OrgStore org_kwargs = OrgStore.get_kwargs_from_user_settings(user_settings) org_kwargs.pop('id', None) org_kwargs['org_version'] = ORG_SETTINGS_VERSION if custom_settings: org_kwargs['agent_settings'] = { 'schema_version': AGENT_SETTINGS_SCHEMA_VERSION, 'llm': { 'model': get_default_llm_model(), 'base_url': get_default_llm_base_url(), }, } return org_kwargs @staticmethod def _has_custom_settings( user_settings: UserSettings, old_user_version: int | None ) -> bool: """Check if user has custom LLM settings that should be preserved. Returns True if user customized either model or base_url. Args: settings: The user's current settings old_user_version: The user's old settings version, if any Returns: True if user has custom settings, False if using old defaults """ persisted_agent_settings = user_settings.agent_settings or {} llm_settings = persisted_agent_settings.get('llm', {}) if isinstance(llm_settings, dict): user_model = llm_settings.get('model') user_base_url = llm_settings.get('base_url') else: user_model = None user_base_url = None user_model = user_model.strip() or None if user_model else None user_base_url = user_base_url.strip() or None if user_base_url else None # Custom base_url = definitely custom settings (BYOK) if user_base_url and user_base_url != LITE_LLM_API_URL: return True # No model set = using defaults if not user_model: return False # Check if model matches old version's default if ( old_user_version and old_user_version <= ORG_SETTINGS_VERSION and old_user_version in PERSONAL_WORKSPACE_VERSION_TO_MODEL ): old_default_base = PERSONAL_WORKSPACE_VERSION_TO_MODEL[old_user_version] user_model_base = user_model.split('/')[-1] if user_model_base == old_default_base: return False # Matches old default return True # Custom model def _is_legacy_value_encrypted(value: str) -> bool: """Check if a legacy value is encrypted by trying to decrypt it""" try: decrypt_legacy_value(value) return True except Exception: return False