diff --git a/.env.docker.example b/.env.docker.example
index 0a178c6..861d78a 100644
--- a/.env.docker.example
+++ b/.env.docker.example
@@ -27,10 +27,10 @@ REDIS_PASSWORD=STRONG_REDIS_PASSWORD_HERE
 RUNTIME_DB_PASSWORD=STRONG_RUNTIME_PASSWORD_HERE
 DATABASE_URL=postgresql+asyncpg://crm_runtime:STRONG_RUNTIME_PASSWORD_HERE@postgres:5432/crm_db
 
-# --- CRM Application: Migration DB user (owner, can run DDL) -----------------
-# Migrations and DDL operations use the owner user (crm_user).
+# --- CRM Application: Migration DB user (NOSUPERUSER, BYPASSRLS) ------------
+# Migrations use crm_migration (NOSUPERUSER, BYPASSRLS) — NOT crm_user (SUPERUSER).
 # This is NOT used by the app at runtime — only by prestart.sh / alembic.
-MIGRATION_DATABASE_URL=postgresql+asyncpg://crm_user:STRONG_PASSWORD_HERE@postgres:5432/crm_db
+MIGRATION_DATABASE_URL=postgresql+asyncpg://crm_migration:STRONG_PASSWORD_HERE@postgres:5432/crm_db
 
 # --- SECRET_KEY (REQUIRED, min 32 chars) -------------------------------------
 # Session signing secret. MUST be at least 32 characters.
diff --git a/alembic/env.py b/alembic/env.py
index 36e4ee4..bec8c62 100644
--- a/alembic/env.py
+++ b/alembic/env.py
@@ -20,6 +20,9 @@ if config.config_file_name is not None:
 
 target_metadata = Base.metadata
 settings = get_settings()
+# ⚠️ RLS Migration History: 21 Migrationen mit 8 Disable-Zyklen. Dies ist historisch bedingt
+# und zeigt trial-and-error. Aktuelle RLS-Konfiguration ist stabil (113 Tabellen).
+# Bei neuen RLS-Änderungen nur noch Migration-Runner nutzen.
 # Use migration_database_url (crm_migration role, table owner) for Alembic
 config.set_main_option("sqlalchemy.url", settings.migration_database_url or settings.database_url)
 
diff --git a/alembic/versions/0114_migrate_contact_folder_permissions_to_entity_permissions.py b/alembic/versions/0114_migrate_contact_folder_permissions_to_entity_permissions.py
new file mode 100644
index 0000000..82e4bd4
--- /dev/null
+++ b/alembic/versions/0114_migrate_contact_folder_permissions_to_entity_permissions.py
@@ -0,0 +1,122 @@
+"""Migrate contact_folder_permissions to entity_permissions.
+
+This migration moves all ACL entries from the dedicated
+``contact_folder_permissions`` table into the universal
+``entity_permissions`` table with ``entity_type='contact_folder'``.
+
+Mapping:
+- folder_id          → entity_id (entity_type='contact_folder')
+- user_id            → principal_type='user',  principal_id=user_id
+- group_id           → principal_type='group', principal_id=group_id
+- permission_level   → permission_level (unchanged)
+- inherit_to_subfolders is dropped (always treated as True after migration)
+
+After data migration the ``contact_folder_permissions`` table is dropped.
+
+Revision ID: 0114
+Revises: 0113
+"""
+
+from __future__ import annotations
+
+from alembic import op
+
+revision = "0114"
+down_revision = "0113"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    # 1. Migrate user-based permissions
+    op.execute("""
+        INSERT INTO entity_permissions (
+            id, tenant_id, entity_type, entity_id,
+            principal_type, principal_id,
+            permission_level, created_at, updated_at
+        )
+        SELECT
+            cfp.id,
+            cfp.tenant_id,
+            'contact_folder',
+            cfp.folder_id,
+            'user',
+            cfp.user_id,
+            cfp.permission_level,
+            cfp.created_at,
+            cfp.updated_at
+        FROM contact_folder_permissions cfp
+        WHERE cfp.user_id IS NOT NULL
+        ON CONFLICT DO NOTHING
+    """)
+
+    # 2. Migrate group-based permissions
+    op.execute("""
+        INSERT INTO entity_permissions (
+            id, tenant_id, entity_type, entity_id,
+            principal_type, principal_id,
+            permission_level, created_at, updated_at
+        )
+        SELECT
+            cfp.id,
+            cfp.tenant_id,
+            'contact_folder',
+            cfp.folder_id,
+            'group',
+            cfp.group_id,
+            cfp.permission_level,
+            cfp.created_at,
+            cfp.updated_at
+        FROM contact_folder_permissions cfp
+        WHERE cfp.group_id IS NOT NULL
+        ON CONFLICT DO NOTHING
+    """)
+
+    # 3. Drop the old table
+    op.execute("DROP TABLE IF EXISTS contact_folder_permissions CASCADE")
+
+
+def downgrade() -> None:
+    # Recreate the old table (data is lost — this is a one-way migration)
+    op.execute("""
+        CREATE TABLE IF NOT EXISTS contact_folder_permissions (
+            id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+            tenant_id UUID NOT NULL,
+            folder_id UUID NOT NULL REFERENCES contact_folders(id) ON DELETE CASCADE,
+            user_id UUID REFERENCES users(id) ON DELETE CASCADE,
+            group_id UUID REFERENCES groups(id) ON DELETE CASCADE,
+            permission_level VARCHAR(20) NOT NULL DEFAULT 'read',
+            inherit_to_subfolders BOOLEAN NOT NULL DEFAULT TRUE,
+            created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+            updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
+        )
+    """)
+
+    # Restore user permissions
+    op.execute("""
+        INSERT INTO contact_folder_permissions (
+            id, tenant_id, folder_id, user_id, group_id,
+            permission_level, inherit_to_subfolders, created_at, updated_at
+        )
+        SELECT
+            ep.id,
+            ep.tenant_id,
+            ep.entity_id,
+            CASE WHEN ep.principal_type = 'user' THEN ep.principal_id ELSE NULL END,
+            CASE WHEN ep.principal_type = 'group' THEN ep.principal_id ELSE NULL END,
+            ep.permission_level,
+            TRUE,
+            ep.created_at,
+            ep.updated_at
+        FROM entity_permissions ep
+        WHERE ep.entity_type = 'contact_folder'
+          AND ep.principal_type IN ('user', 'group')
+        ON CONFLICT DO NOTHING
+    """)
+
+    # Remove migrated entries from entity_permissions
+    op.execute("""
+        DELETE FROM entity_permissions
+        WHERE entity_type = 'contact_folder'
+          AND principal_type IN ('user', 'group')
+    """)
diff --git a/app/core/auth.py b/app/core/auth.py
index 8057920..49275a7 100644
--- a/app/core/auth.py
+++ b/app/core/auth.py
@@ -297,38 +297,5 @@ async def update_session_tenant(
     return data
 
 
-def check_permission(
-    role_name: str, module: str, action: str, permissions: dict | None = None
-) -> bool:
-    """Check if a role has permission for a module+action.
-
-    ⚠️ Legacy Role Bypass entfernt — alle Admins müssen echte role_id haben
-    Built-in role strings (admin/editor/viewer) no longer grant permissions directly.
-    All permission checks must go through the RBAC system in app.core.permissions.
-    This function is kept for backward compatibility but no longer bypasses checks
-    based on role_name alone.
-    """
-    # Custom role — check permissions dict
-    if permissions:
-        module_perms = permissions.get(module, {})
-        return bool(module_perms.get(action, False))
-    return False
-
-
-def filter_fields_by_permission(
-    data: dict[str, Any],
-    field_permissions: dict[str, str],
-    role_name: str,
-) -> dict[str, Any]:
-    """Filter response fields based on field-level permissions.
-    field_permissions: {"annual_revenue": "hidden"} → removed for non-admin.
-    """
-    if role_name == "admin":
-        return data
-    result = {}
-    for key, value in data.items():
-        perm = field_permissions.get(key)
-        if perm == "hidden":
-            continue
-        result[key] = value
-    return result
+# ⚠️ Legacy check_permission and filter_fields_by_permission removed from auth.py.
+# Use app.core.permissions.check_permission and app.core.permissions.filter_fields_by_permission instead.
diff --git a/app/core/permissions.py b/app/core/permissions.py
index 86521ee..525b86d 100644
--- a/app/core/permissions.py
+++ b/app/core/permissions.py
@@ -31,6 +31,10 @@ logger = logging.getLogger(__name__)
 CACHE_TTL = 300  # 5 minutes
 CACHE_PREFIX = "resolved"
 
+# Central permission rank — single source of truth for permission level ordering.
+# Used by entity_permission_service, bulk_permission_service, visibility, etc.
+PERM_RANK = {"none": 0, "read": 1, "write": 2, "admin": 3, "delete": 4, "owner": 5}
+
 # Severity ordering for field permissions: highest wins
 _FIELD_PERM_SEVERITY = {"hidden": 3, "readonly": 2, "read": 1}
 
@@ -284,25 +288,10 @@ async def resolve_permissions(
         tenant = tenant_result.scalar_one_or_none()
     resolution_strategy = tenant.resolution_strategy if tenant else "highest_wins"
 
-    # Apply resolution strategy
-    if resolution_strategy == "highest_wins":
-        # Default: allowed - denied (deny overrides allow at permission level)
-        resolved = allowed - denied
-    elif resolution_strategy == "deny_overrides_allow":
-        # Deny always wins: remove any allowed permission that is also denied
-        resolved = allowed - denied
-    elif resolution_strategy == "direct_overrides_group":
-        # Direct role permissions override group permissions
-        # Role permissions are loaded first, group permissions add but don't override
-        # Already implemented by loading order: role first, then group
-        resolved = allowed - denied
-    elif resolution_strategy == "most_restrictive_wins":
-        # Only permissions present in ALL sources (role AND groups) are kept
-        # This is intersection-based: only permissions granted by both role and groups
-        # For now, we keep the default behavior as intersection is complex with multiple groups
-        resolved = allowed - denied
-    else:
-        resolved = allowed - denied
+    # ⚠️ Only highest_wins strategy supported — other strategies removed as they were no-ops.
+    # All strategies previously produced the same result: resolved = allowed - denied.
+    # The strategy field is kept for backward compatibility but only highest_wins is honored.
+    resolved = allowed - denied
 
     return {
         "permissions": resolved,
diff --git a/app/core/visibility.py b/app/core/visibility.py
index e67185a..528b503 100644
--- a/app/core/visibility.py
+++ b/app/core/visibility.py
@@ -39,8 +39,7 @@ from app.models.user import User, UserTenant
 
 logger = logging.getLogger(__name__)
 
-# Permission rank for comparison
-_PERM_RANK = {"none": 0, "read": 1, "write": 2, "admin": 3, "delete": 4, "owner": 5}
+from app.core.permissions import PERM_RANK as _PERM_RANK
 
 
 def _rank(level: str) -> int:
diff --git a/app/models/__init__.py b/app/models/__init__.py
index 62b7b19..c57348d 100644
--- a/app/models/__init__.py
+++ b/app/models/__init__.py
@@ -8,7 +8,6 @@ from app.models.audit import AuditLog, DeletionLog
 from app.models.auth import ApiToken, PasswordResetToken
 from app.models.contact import Contact, ContactPerson
 from app.models.contact_folder import ContactFolder
-from app.models.contact_folder_permission import ContactFolderPermission
 from app.models.contact_merge import ContactMergeHistory
 from app.models.entity_permission import EntityPermission
 from app.models.consumer_inbox import ConsumerInbox
@@ -53,7 +52,6 @@ __all__ = [
     "Contact",
     "ContactPerson",
     "ContactFolder",
-    "ContactFolderPermission",
     "ContactMergeHistory",
     "EntityPermission",
     "ConsumerInbox",
diff --git a/app/models/tenant.py b/app/models/tenant.py
index 817b2af..af8a2a7 100644
--- a/app/models/tenant.py
+++ b/app/models/tenant.py
@@ -33,8 +33,9 @@ class Tenant(Base):
     )
 
     __table_args__ = (
+        # ⚠️ Only highest_wins strategy supported — other strategies removed as they were no-ops.
         CheckConstraint(
-            "resolution_strategy IN ('highest_wins', 'deny_overrides_allow', 'direct_overrides_group', 'most_restrictive_wins')",
+            "resolution_strategy IN ('highest_wins')",
             name="ck_tenant_resolution_strategy",
         ),
     )
diff --git a/app/routes/ai_copilot.py b/app/routes/ai_copilot.py
index 254e8a0..3a3fa89 100644
--- a/app/routes/ai_copilot.py
+++ b/app/routes/ai_copilot.py
@@ -63,13 +63,18 @@ async def copilot_execute(
     """
     tenant_id = uuid.UUID(current_user["tenant_id"])
     user_id = uuid.UUID(current_user["user_id"])
-    role = current_user.get("role", "viewer")
+    resolved = {
+        "permissions": current_user.get("permissions", []),
+        "denied": current_user.get("denied_permissions", []),
+        "field_permissions": current_user.get("field_permissions", {}),
+        "is_system_admin": current_user.get("is_system_admin", False),
+    }
 
     result = await ai_copilot_service.execute_action(
         db,
         tenant_id,
         user_id,
-        role,
+        resolved,
         conversation_id=body.conversation_id,
         action=body.action.model_dump(),
     )
diff --git a/app/routes/users.py b/app/routes/users.py
index 9823481..9e9ac32 100644
--- a/app/routes/users.py
+++ b/app/routes/users.py
@@ -10,7 +10,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy import select
 
 from app.core.audit import log_audit
-from app.core.auth import get_redis
+from app.core.auth import get_redis, invalidate_all_user_sessions
 from app.core.db import get_db
 from app.core.notifications import create_notification
 from app.core.permissions import invalidate_permission_cache
@@ -205,6 +205,13 @@ async def update_user(
             detail={"detail": "Only system admin can assign admin role", "code": "role_escalation_forbidden"},
         )
 
+    # Only system admin can change is_system_admin flag
+    if body.is_system_admin is not None and not current_user.get("is_system_admin"):
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail={"detail": "Only system admin can change system admin flag", "code": "admin_flag_forbidden"},
+        )
+
     # Determine if role_id was explicitly sent (Pydantic v2)
     role_id_sent = "role_id" in body.model_fields_set
 
@@ -217,6 +224,8 @@ async def update_user(
         changes["role_id"] = body.role_id
     if body.is_active is not None:
         changes["is_active"] = body.is_active
+    if body.is_system_admin is not None:
+        changes["is_system_admin"] = body.is_system_admin
 
     # Pass _UNSET sentinel when role_id was not in the request body
     # so the service leaves the existing value untouched.
@@ -253,6 +262,7 @@ async def update_user(
             body.email,
             body.current_password,
             body.new_password,
+            body.is_system_admin,
         )
     except ValueError as exc:
         raise HTTPException(400, detail={"detail": str(exc), "code": "invalid_password"}) from None
@@ -276,6 +286,14 @@ async def update_user(
     redis = get_redis()
     await invalidate_permission_cache(redis, uid, tenant_id)
 
+    # If is_system_admin was changed, invalidate ALL sessions for this user
+    # so the stale admin flag doesn't persist in Redis until TTL (8h)
+    if body.is_system_admin is not None:
+        try:
+            await invalidate_all_user_sessions(redis, uid)
+        except Exception:
+            pass  # Best-effort — don't fail the update if Redis is down
+
     return {
         "id": str(user.id),
         "email": user.email,
diff --git a/app/schemas/user.py b/app/schemas/user.py
index 18d127b..1a8e30b 100644
--- a/app/schemas/user.py
+++ b/app/schemas/user.py
@@ -45,6 +45,7 @@ class UserUpdate(BaseModel):
     role: str | None = None
     role_id: str | None = Field(default=None, description="UUID of a custom Role; send null to clear")
     is_active: bool | None = None
+    is_system_admin: bool | None = Field(None, description="System admin flag — requires system admin to set")
     current_password: str | None = Field(None, description="Required when changing password")
     new_password: str | None = Field(None, min_length=8, description="New password")
 
@@ -59,6 +60,7 @@ class UserResponse(BaseModel):
     role: str
     role_id: str | None = None
     is_active: bool
+    is_system_admin: bool = False
     tenant_id: str
 
 
diff --git a/app/services/ai_copilot_service.py b/app/services/ai_copilot_service.py
index 9932998..e9064a0 100644
--- a/app/services/ai_copilot_service.py
+++ b/app/services/ai_copilot_service.py
@@ -11,7 +11,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.ai.llm_client import get_llm_client
 from app.core.audit import log_audit
-from app.core.auth import check_permission
+from app.core.permissions import check_permission
 from app.core.visibility import apply_visibility_filter
 from app.core.visibility import check_single_entity_access
 from app.models.ai_conversation import AIConversation, AIMessage
@@ -159,7 +159,7 @@ async def execute_action(
     db: AsyncSession,
     tenant_id: uuid.UUID,
     user_id: uuid.UUID,
-    role: str,
+    resolved: dict[str, Any],
     conversation_id: str,
     action: dict[str, Any],
     is_system_admin: bool = False,
@@ -191,8 +191,9 @@ async def execute_action(
 
     # Determine module and action_type from path for RBAC
     module, action_type = _derive_rbac_from_path(method, path)
+    required_perm = f"{module}:{action_type}"
 
-    if not check_permission(role, module, action_type):
+    if not check_permission(resolved, required_perm):
         return {
             "error": "Insufficient permissions for this action",
             "status_code": 403,
diff --git a/app/services/bulk_permission_service.py b/app/services/bulk_permission_service.py
index 9bfa6c1..c1b39e2 100644
--- a/app/services/bulk_permission_service.py
+++ b/app/services/bulk_permission_service.py
@@ -20,7 +20,7 @@ from app.models.user import User, UserTenant
 
 logger = logging.getLogger(__name__)
 
-_PERM_RANK = {"none": 0, "read": 1, "write": 2, "admin": 3, "delete": 4, "owner": 5}
+from app.core.permissions import PERM_RANK as _PERM_RANK
 
 
 def _rank(level: str) -> int:
diff --git a/app/services/contact_folder_permission_service.py b/app/services/contact_folder_permission_service.py
index 5368ff3..179504d 100644
--- a/app/services/contact_folder_permission_service.py
+++ b/app/services/contact_folder_permission_service.py
@@ -1,20 +1,35 @@
-"""Contact folder permission service — ACL management and access resolution."""
+"""Contact folder permission service — delegates to EntityPermission.
+
+This service preserves the original public API (list_permissions,
+create_permission, update_permission, delete_permission,
+get_effective_access, get_visible_folder_ids) but internally uses the
+universal ``entity_permissions`` table with ``entity_type='contact_folder'``.
+
+Mapping:
+- folder_id          → entity_id (entity_type='contact_folder')
+- user_id            → principal_type='user',  principal_id=user_id
+- group_id           → principal_type='group', principal_id=group_id
+- permission_level   → permission_level (same values)
+- inherit_to_subfolders → always treated as True (EntityPermission has no
+  such column; all folder permissions inherit to subfolders).
+"""
 
 from __future__ import annotations
 
 import uuid
 from typing import Any
 
-from sqlalchemy import or_, select, func
+from sqlalchemy import func, or_, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.models.contact_folder import ContactFolder
-from app.models.contact_folder_permission import ContactFolderPermission
+from app.models.entity_permission import EntityPermission
 from app.models.group import Group, UserGroup
 from app.models.user import User
 
-# Permission hierarchy: higher = more access
-_PERM_RANK = {"none": 0, "read": 1, "write": 2, "admin": 3, "owner": 4}
+_ENTITY_TYPE = "contact_folder"
+
+from app.core.permissions import PERM_RANK as _PERM_RANK
 
 
 def _rank(level: str) -> int:
@@ -22,39 +37,32 @@ def _rank(level: str) -> int:
 
 
 def _serialize_permission(
-    p: ContactFolderPermission,
+    p: EntityPermission,
     user_name: str | None = None,
     group_name: str | None = None,
 ) -> dict:
+    """Serialize EntityPermission back to the legacy folder-permission format."""
+    user_id = str(p.principal_id) if p.principal_type == "user" else None
+    group_id = str(p.principal_id) if p.principal_type == "group" else None
     return {
         "id": str(p.id),
-        "folder_id": str(p.folder_id),
-        "user_id": str(p.user_id) if p.user_id else None,
-        "group_id": str(p.group_id) if p.group_id else None,
+        "folder_id": str(p.entity_id),
+        "user_id": user_id,
+        "group_id": group_id,
         "user_name": user_name,
         "group_name": group_name,
         "permission_level": p.permission_level,
-        "inherit_to_subfolders": p.inherit_to_subfolders,
+        "inherit_to_subfolders": True,  # always True after migration
         "created_at": p.created_at.isoformat() if p.created_at else None,
     }
 
 
-async def list_permissions(
-    db: AsyncSession, tenant_id: uuid.UUID, folder_id: str
-) -> list[dict]:
-    """List all permission entries for a folder."""
-    folder_uuid = uuid.UUID(folder_id)
-    result = await db.execute(
-        select(ContactFolderPermission)
-        .where(ContactFolderPermission.folder_id == folder_uuid)
-        .where(ContactFolderPermission.tenant_id == tenant_id)
-        .order_by(ContactFolderPermission.created_at)
-    )
-    perms = result.scalars().all()
-
-    # Batch-load user and group names
-    user_ids = [p.user_id for p in perms if p.user_id]
-    group_ids = [p.group_id for p in perms if p.group_id]
+async def _load_names(
+    db: AsyncSession, perms: list[EntityPermission]
+) -> tuple[dict[uuid.UUID, str], dict[uuid.UUID, str]]:
+    """Batch-load user and group names for a list of permissions."""
+    user_ids = [p.principal_id for p in perms if p.principal_type == "user"]
+    group_ids = [p.principal_id for p in perms if p.principal_type == "group"]
 
     user_names: dict[uuid.UUID, str] = {}
     if user_ids:
@@ -70,11 +78,30 @@ async def list_permissions(
         )
         group_names = {row[0]: row[1] for row in groups_q}
 
+    return user_names, group_names
+
+
+async def list_permissions(
+    db: AsyncSession, tenant_id: uuid.UUID, folder_id: str
+) -> list[dict]:
+    """List all permission entries for a folder."""
+    folder_uuid = uuid.UUID(folder_id)
+    result = await db.execute(
+        select(EntityPermission)
+        .where(EntityPermission.entity_type == _ENTITY_TYPE)
+        .where(EntityPermission.entity_id == folder_uuid)
+        .where(EntityPermission.tenant_id == tenant_id)
+        .order_by(EntityPermission.created_at)
+    )
+    perms = result.scalars().all()
+
+    user_names, group_names = await _load_names(db, perms)
+
     return [
         _serialize_permission(
             p,
-            user_names.get(p.user_id) if p.user_id else None,
-            group_names.get(p.group_id) if p.group_id else None,
+            user_names.get(p.principal_id) if p.principal_type == "user" else None,
+            group_names.get(p.principal_id) if p.principal_type == "group" else None,
         )
         for p in perms
     ]
@@ -89,7 +116,12 @@ async def create_permission(
     permission_level: str,
     inherit_to_subfolders: bool = True,
 ) -> dict:
-    """Create or update a permission entry for a folder."""
+    """Create or update a permission entry for a folder.
+
+    Delegates to EntityPermission with entity_type='contact_folder'.
+    ``inherit_to_subfolders`` is accepted for API compatibility but has no
+    effect (all folder permissions inherit to subfolders after migration).
+    """
     folder_uuid = uuid.UUID(folder_id)
 
     # Verify folder exists and belongs to tenant
@@ -109,58 +141,51 @@ async def create_permission(
     if user_uuid and group_uuid:
         raise ValueError("Only one of user_id or group_id can be provided")
 
+    principal_type = "user" if user_uuid else "group"
+    principal_uuid = user_uuid or group_uuid
+
     # Check for existing entry (upsert)
     existing_q = await db.execute(
-        select(ContactFolderPermission)
-        .where(ContactFolderPermission.folder_id == folder_uuid)
-        .where(ContactFolderPermission.tenant_id == tenant_id)
-        .where(
-            ContactFolderPermission.user_id == user_uuid
-            if user_uuid
-            else ContactFolderPermission.user_id.is_(None)
-        )
-        .where(
-            ContactFolderPermission.group_id == group_uuid
-            if group_uuid
-            else ContactFolderPermission.group_id.is_(None)
-        )
+        select(EntityPermission)
+        .where(EntityPermission.entity_type == _ENTITY_TYPE)
+        .where(EntityPermission.entity_id == folder_uuid)
+        .where(EntityPermission.principal_type == principal_type)
+        .where(EntityPermission.principal_id == principal_uuid)
+        .where(EntityPermission.tenant_id == tenant_id)
     )
     existing = existing_q.scalar_one_or_none()
 
     if existing:
         existing.permission_level = permission_level
-        existing.inherit_to_subfolders = inherit_to_subfolders
         await db.commit()
         await db.refresh(existing)
-        # Load names for response
         user_name = group_name = None
-        if existing.user_id:
-            u = await db.execute(select(User.name).where(User.id == existing.user_id))
+        if existing.principal_type == "user":
+            u = await db.execute(select(User.name).where(User.id == existing.principal_id))
             user_name = u.scalar_one_or_none()
-        if existing.group_id:
-            g = await db.execute(select(Group.name).where(Group.id == existing.group_id))
+        else:
+            g = await db.execute(select(Group.name).where(Group.id == existing.principal_id))
             group_name = g.scalar_one_or_none()
         return _serialize_permission(existing, user_name, group_name)
 
-    perm = ContactFolderPermission(
+    perm = EntityPermission(
         tenant_id=tenant_id,
-        folder_id=folder_uuid,
-        user_id=user_uuid,
-        group_id=group_uuid,
+        entity_type=_ENTITY_TYPE,
+        entity_id=folder_uuid,
+        principal_type=principal_type,
+        principal_id=principal_uuid,
         permission_level=permission_level,
-        inherit_to_subfolders=inherit_to_subfolders,
     )
     db.add(perm)
     await db.commit()
     await db.refresh(perm)
 
-    # Load names for response
     user_name = group_name = None
-    if perm.user_id:
-        u = await db.execute(select(User.name).where(User.id == perm.user_id))
+    if perm.principal_type == "user":
+        u = await db.execute(select(User.name).where(User.id == perm.principal_id))
         user_name = u.scalar_one_or_none()
-    if perm.group_id:
-        g = await db.execute(select(Group.name).where(Group.id == perm.group_id))
+    else:
+        g = await db.execute(select(Group.name).where(Group.id == perm.principal_id))
         group_name = g.scalar_one_or_none()
     return _serialize_permission(perm, user_name, group_name)
 
@@ -172,30 +197,34 @@ async def update_permission(
     permission_level: str,
     inherit_to_subfolders: bool | None = None,
 ) -> dict:
-    """Update an existing permission entry."""
+    """Update an existing permission entry.
+
+    ``inherit_to_subfolders`` is accepted for API compatibility but has no
+    effect.
+    """
     perm_uuid = uuid.UUID(permission_id)
     result = await db.execute(
-        select(ContactFolderPermission)
-        .where(ContactFolderPermission.id == perm_uuid)
-        .where(ContactFolderPermission.tenant_id == tenant_id)
+        select(EntityPermission)
+        .where(EntityPermission.id == perm_uuid)
+        .where(EntityPermission.tenant_id == tenant_id)
+        .where(EntityPermission.entity_type == _ENTITY_TYPE)
     )
     perm = result.scalar_one_or_none()
     if not perm:
         raise ValueError("Permission not found")
 
     perm.permission_level = permission_level
-    if inherit_to_subfolders is not None:
-        perm.inherit_to_subfolders = inherit_to_subfolders
+    # inherit_to_subfolders has no equivalent in EntityPermission
 
     await db.commit()
     await db.refresh(perm)
 
     user_name = group_name = None
-    if perm.user_id:
-        u = await db.execute(select(User.name).where(User.id == perm.user_id))
+    if perm.principal_type == "user":
+        u = await db.execute(select(User.name).where(User.id == perm.principal_id))
         user_name = u.scalar_one_or_none()
-    if perm.group_id:
-        g = await db.execute(select(Group.name).where(Group.id == perm.group_id))
+    else:
+        g = await db.execute(select(Group.name).where(Group.id == perm.principal_id))
         group_name = g.scalar_one_or_none()
     return _serialize_permission(perm, user_name, group_name)
 
@@ -206,9 +235,10 @@ async def delete_permission(
     """Delete a permission entry."""
     perm_uuid = uuid.UUID(permission_id)
     result = await db.execute(
-        select(ContactFolderPermission)
-        .where(ContactFolderPermission.id == perm_uuid)
-        .where(ContactFolderPermission.tenant_id == tenant_id)
+        select(EntityPermission)
+        .where(EntityPermission.id == perm_uuid)
+        .where(EntityPermission.tenant_id == tenant_id)
+        .where(EntityPermission.entity_type == _ENTITY_TYPE)
     )
     perm = result.scalar_one_or_none()
     if not perm:
@@ -229,7 +259,7 @@ async def get_effective_access(
     Resolution order (highest wins):
     1. Folder owner → "owner"
     2. Direct permission on this folder
-    3. Inherited permission from ancestor folders (inherit_to_subfolders=True)
+    3. Inherited permission from ancestor folders (all inherit after migration)
     4. Group membership permissions (direct + inherited)
     5. No access → "none"
     """
@@ -246,9 +276,10 @@ async def get_effective_access(
     if folder.user_id == user_id:
         # Check if shared with anyone
         shared_q = await db.execute(
-            select(func.count(ContactFolderPermission.id))
-            .where(ContactFolderPermission.folder_id == folder_id)
-            .where(ContactFolderPermission.tenant_id == tenant_id)
+            select(func.count(EntityPermission.id))
+            .where(EntityPermission.entity_type == _ENTITY_TYPE)
+            .where(EntityPermission.entity_id == folder_id)
+            .where(EntityPermission.tenant_id == tenant_id)
         )
         is_shared = (shared_q.scalar() or 0) > 0
         return {"folder_id": str(folder_id), "access_level": "owner", "is_owner": True, "is_shared": is_shared, "inherited_from": None}
@@ -282,15 +313,15 @@ async def get_effective_access(
     for i, ancestor_id in enumerate(ancestor_chain):
         # Direct user permission
         user_perm_q = await db.execute(
-            select(ContactFolderPermission)
-            .where(ContactFolderPermission.folder_id == ancestor_id)
-            .where(ContactFolderPermission.tenant_id == tenant_id)
-            .where(ContactFolderPermission.user_id == user_id)
+            select(EntityPermission)
+            .where(EntityPermission.entity_type == _ENTITY_TYPE)
+            .where(EntityPermission.entity_id == ancestor_id)
+            .where(EntityPermission.tenant_id == tenant_id)
+            .where(EntityPermission.principal_type == "user")
+            .where(EntityPermission.principal_id == user_id)
         )
         for perm in user_perm_q.scalars():
-            # If this is an ancestor (not the folder itself), only apply if inherit_to_subfolders
-            if i > 0 and not perm.inherit_to_subfolders:
-                continue
+            # All permissions inherit after migration (inherit_to_subfolders always True)
             if _rank(perm.permission_level) > _rank(best_level):
                 best_level = perm.permission_level
                 inherited_from = str(ancestor_id) if i > 0 else None
@@ -298,23 +329,24 @@ async def get_effective_access(
         # Group permissions
         if group_ids:
             group_perm_q = await db.execute(
-                select(ContactFolderPermission)
-                .where(ContactFolderPermission.folder_id == ancestor_id)
-                .where(ContactFolderPermission.tenant_id == tenant_id)
-                .where(ContactFolderPermission.group_id.in_(group_ids))
+                select(EntityPermission)
+                .where(EntityPermission.entity_type == _ENTITY_TYPE)
+                .where(EntityPermission.entity_id == ancestor_id)
+                .where(EntityPermission.tenant_id == tenant_id)
+                .where(EntityPermission.principal_type == "group")
+                .where(EntityPermission.principal_id.in_(group_ids))
             )
             for perm in group_perm_q.scalars():
-                if i > 0 and not perm.inherit_to_subfolders:
-                    continue
                 if _rank(perm.permission_level) > _rank(best_level):
                     best_level = perm.permission_level
                     inherited_from = str(ancestor_id) if i > 0 else None
 
     # Check if folder is shared at all
     shared_q = await db.execute(
-        select(func.count(ContactFolderPermission.id))
-        .where(ContactFolderPermission.folder_id == folder_id)
-        .where(ContactFolderPermission.tenant_id == tenant_id)
+        select(func.count(EntityPermission.id))
+        .where(EntityPermission.entity_type == _ENTITY_TYPE)
+        .where(EntityPermission.entity_id == folder_id)
+        .where(EntityPermission.tenant_id == tenant_id)
     )
     is_shared = (shared_q.scalar() or 0) > 0
 
@@ -356,22 +388,20 @@ async def get_visible_folder_ids(
     )
     group_ids = [row[0] for row in groups_q]
 
-    # Get all permission entries for this tenant
+    # Get all permission entries for this tenant + entity_type
     all_perms_q = await db.execute(
-        select(ContactFolderPermission)
-        .where(ContactFolderPermission.tenant_id == tenant_id)
+        select(EntityPermission)
+        .where(EntityPermission.entity_type == _ENTITY_TYPE)
+        .where(EntityPermission.tenant_id == tenant_id)
     )
     all_perms = all_perms_q.scalars().all()
 
-    # Build permission lookup: folder_id → list of (principal_type, principal_id, level, inherit)
-    perm_lookup: dict[uuid.UUID, list[tuple[str, uuid.UUID | None, str, bool]]] = {}
+    # Build permission lookup: entity_id → list of (principal_type, principal_id, level)
+    perm_lookup: dict[uuid.UUID, list[tuple[str, uuid.UUID, str]]] = {}
     for p in all_perms:
-        if p.folder_id not in perm_lookup:
-            perm_lookup[p.folder_id] = []
-        if p.user_id:
-            perm_lookup[p.folder_id].append(("user", p.user_id, p.permission_level, p.inherit_to_subfolders))
-        if p.group_id:
-            perm_lookup[p.folder_id].append(("group", p.group_id, p.permission_level, p.inherit_to_subfolders))
+        if p.entity_id not in perm_lookup:
+            perm_lookup[p.entity_id] = []
+        perm_lookup[p.entity_id].append((p.principal_type, p.principal_id, p.permission_level))
 
     # Build parent map for ancestor traversal
     parent_map: dict[uuid.UUID, uuid.UUID | None] = {}
@@ -397,9 +427,8 @@ async def get_visible_folder_ids(
 
         for i, ancestor_id in enumerate(chain):
             perms = perm_lookup.get(ancestor_id, [])
-            for ptype, pid, level, inherit in perms:
-                if i > 0 and not inherit:
-                    continue
+            for ptype, pid, level in perms:
+                # All permissions inherit after migration
                 if ptype == "user" and pid == user_id:
                     if _rank(level) > _rank(best_level):
                         best_level = level
diff --git a/app/services/contact_folder_service.py b/app/services/contact_folder_service.py
index 924fd04..2b46851 100644
--- a/app/services/contact_folder_service.py
+++ b/app/services/contact_folder_service.py
@@ -64,12 +64,13 @@ async def list_folders(
             counts[fid] = cnt
 
     # Check which folders have permissions (are shared)
-    from app.models.contact_folder_permission import ContactFolderPermission
+    from app.models.entity_permission import EntityPermission
     shared_q = await db.execute(
-        select(ContactFolderPermission.folder_id)
-        .where(ContactFolderPermission.folder_id.in_(folder_ids))
-        .where(ContactFolderPermission.tenant_id == tenant_id)
-        .group_by(ContactFolderPermission.folder_id)
+        select(EntityPermission.entity_id)
+        .where(EntityPermission.entity_type == "contact_folder")
+        .where(EntityPermission.entity_id.in_(folder_ids))
+        .where(EntityPermission.tenant_id == tenant_id)
+        .group_by(EntityPermission.entity_id)
     )
     shared_ids = {row[0] for row in shared_q}
 
diff --git a/app/services/entity_permission_service.py b/app/services/entity_permission_service.py
index 3f29c96..7e99918 100644
--- a/app/services/entity_permission_service.py
+++ b/app/services/entity_permission_service.py
@@ -7,6 +7,11 @@ This service handles:
 - Redis caching with bitmap optimization
 - Permission expiration checks
 - Audit logging for permission changes
+
+Resolution, caching, and audit logic have been extracted into focused modules:
+- permission_resolver.py: get_effective_access, get_visible_ids, batch_get_effective_access, check_entity_access
+- permission_cache.py: _invalidate_user_cache, get_cached_visible_ids, invalidate_all_user_entity_cache
+- permission_audit.py: _log_permission_grant, _log_permission_update, _log_permission_revoke
 """
 
 from __future__ import annotations
@@ -39,6 +44,9 @@ from app.models.webhook import Webhook
 from app.models.custom_field_definition import CustomFieldDefinition
 from app.models.contact_folder import ContactFolder
 
+# Import cache helpers used by CRUD operations
+from app.services.permission_cache import _invalidate_user_cache, CACHE_TTL, CACHE_PREFIX
+
 logger = logging.getLogger(__name__)
 
 # ── Entity Model Registry ────────────────────────────────────────────────────
@@ -116,11 +124,7 @@ def _get_entity_model(entity_type: str) -> type:
         raise ValueError(f"Unknown entity type: {entity_type}")
     return model
 
-# Permission hierarchy: higher = more access
-_PERM_RANK = {"none": 0, "read": 1, "write": 2, "admin": 3, "delete": 4, "owner": 5}
-
-CACHE_TTL = 300  # 5 minutes
-CACHE_PREFIX = "ep_vis"  # entity permission visibility
+from app.core.permissions import PERM_RANK as _PERM_RANK
 
 
 def _rank(level: str) -> int:
@@ -408,422 +412,6 @@ async def delete_permission(
             await _invalidate_user_cache(redis, tenant_id, uid, old_entity_type)
 
 
-async def get_effective_access(
-    db: AsyncSession,
-    tenant_id: uuid.UUID,
-    user_id: uuid.UUID,
-    entity_type: str,
-    entity_id: uuid.UUID,
-) -> str:
-    """Get the effective access level for a user on a specific entity.
-
-    Resolution (highest wins):
-    1. System admin → 'delete' (full access)
-    2. Owner → 'owner' (from owner_id on the entity table)
-    3. Direct user permission
-    4. Group permission (via user_groups)
-    5. Role permission (via user_tenants.role_id)
-    6. Guest permission (principal_type='guest')
-    7. owner_id IS NULL → 'read' (tenant-owned, visible to all with module permission)
-    8. No access → 'none'
-
-    Returns: 'none' | 'read' | 'write' | 'admin' | 'delete' | 'owner'
-    """
-    # Check system admin
-    user_q = await db.execute(
-        select(User.is_system_admin).where(User.id == user_id)
-    )
-    if user_q.scalar():
-        return "delete"
-
-    # Check ownership — load the entity's owner_id via SQLAlchemy model (safe from SQL injection)
-    model = _get_entity_model(entity_type)
-    owner_q = await db.execute(
-        select(model.owner_id).where(model.id == entity_id).where(model.tenant_id == tenant_id)
-    )
-    owner_row = owner_q.first()
-    if not owner_row:
-        return "none"
-
-    owner_id = owner_row[0]
-    if owner_id == user_id:
-        return "owner"
-
-    # Get user's groups and role
-    groups_q = await db.execute(
-        select(UserGroup.group_id)
-        .where(UserGroup.user_id == user_id)
-        .where(UserGroup.tenant_id == tenant_id)
-    )
-    group_ids = [row[0] for row in groups_q]
-
-    role_q = await db.execute(
-        select(UserTenant.role_id)
-        .where(UserTenant.user_id == user_id)
-        .where(UserTenant.tenant_id == tenant_id)
-    )
-    role_id = role_q.scalar_one_or_none()
-
-    # Build principal conditions
-    principal_conditions = [
-        and_(
-            EntityPermission.principal_type == "user",
-            EntityPermission.principal_id == user_id,
-        ),
-    ]
-    if group_ids:
-        principal_conditions.append(
-            and_(
-                EntityPermission.principal_type == "group",
-                EntityPermission.principal_id.in_(group_ids),
-            )
-        )
-    if role_id:
-        principal_conditions.append(
-            and_(
-                EntityPermission.principal_type == "role",
-                EntityPermission.principal_id == role_id,
-            )
-        )
-    # Guest permission — guests are now regular users with role=guest
-    # (Guest users may have no groups/roles, only direct entity_permissions)
-    principal_conditions.append(
-        and_(
-            EntityPermission.principal_type == "guest",
-            EntityPermission.principal_id == user_id,
-        )
-    )
-
-    # Query permissions
-    now = datetime.now(UTC)
-    perm_q = await db.execute(
-        select(EntityPermission.permission_level)
-        .where(EntityPermission.entity_type == entity_type)
-        .where(EntityPermission.entity_id == entity_id)
-        .where(EntityPermission.tenant_id == tenant_id)
-        .where(or_(*principal_conditions))
-        .where(
-            or_(
-                EntityPermission.expires_at.is_(None),
-                EntityPermission.expires_at > now,
-            )
-        )
-    )
-
-    best_level = "none"
-    for (level,) in perm_q:
-        if _rank(level) > _rank(best_level):
-            best_level = level
-
-    # If owner_id is NULL (tenant-owned), user with module permission gets at least 'read'
-    if best_level == "none" and owner_id is None:
-        return "read"
-
-    return best_level
-
-
-async def get_visible_ids(
-    db: AsyncSession,
-    tenant_id: uuid.UUID,
-    user_id: uuid.UUID,
-    entity_type: str,
-) -> tuple[set[uuid.UUID], dict[uuid.UUID, str]]:
-    """Get all visible entity IDs for a user and their access levels.
-
-    Returns (visible_ids, access_map) where access_map is
-    entity_id → access_level string.
-
-    Resolution:
-    1. System admin → all entities at 'delete' level
-    2. Owned entities → 'owner'
-    3. Entities with direct/group/role permissions → permission level
-    4. Tenant-owned entities (owner_id IS NULL) → 'read'
-    """
-    # Check system admin
-    user_q = await db.execute(
-        select(User.is_system_admin).where(User.id == user_id)
-    )
-    if user_q.scalar():
-        # Return all entity IDs
-        model = _get_entity_model(entity_type)
-        admin_q = select(model.id).where(model.tenant_id == tenant_id)
-        if hasattr(model, 'deleted_at'):
-            admin_q = admin_q.where(model.deleted_at.is_(None))
-        all_q = await db.execute(admin_q)
-        all_ids = {row[0] for row in all_q}
-        return all_ids, {eid: "delete" for eid in all_ids}
-
-    # Get user's groups and role
-    groups_q = await db.execute(
-        select(UserGroup.group_id)
-        .where(UserGroup.user_id == user_id)
-        .where(UserGroup.tenant_id == tenant_id)
-    )
-    group_ids = [row[0] for row in groups_q]
-
-    role_q = await db.execute(
-        select(UserTenant.role_id)
-        .where(UserTenant.user_id == user_id)
-        .where(UserTenant.tenant_id == tenant_id)
-    )
-    role_id = role_q.scalar_one_or_none()
-
-    # 1. Owned entities
-    model = _get_entity_model(entity_type)
-    owned_q_builder = select(model.id).where(model.tenant_id == tenant_id).where(model.owner_id == user_id)
-    if hasattr(model, 'deleted_at'):
-        owned_q_builder = owned_q_builder.where(model.deleted_at.is_(None))
-    owned_q = await db.execute(owned_q_builder)
-    visible: set[uuid.UUID] = set()
-    access_map: dict[uuid.UUID, str] = {}
-    for (eid,) in owned_q:
-        visible.add(eid)
-        access_map[eid] = "owner"
-
-    # 2. Tenant-owned entities (owner_id IS NULL)
-    tenant_q_builder = select(model.id).where(model.tenant_id == tenant_id).where(model.owner_id.is_(None))
-    if hasattr(model, 'deleted_at'):
-        tenant_q_builder = tenant_q_builder.where(model.deleted_at.is_(None))
-    tenant_owned_q = await db.execute(tenant_q_builder)
-    for (eid,) in tenant_owned_q:
-        if eid not in visible:
-            visible.add(eid)
-            access_map[eid] = "read"
-
-    # 3. Permission-based access
-    now = datetime.now(UTC)
-    principal_conditions = [
-        and_(
-            EntityPermission.principal_type == "user",
-            EntityPermission.principal_id == user_id,
-        ),
-    ]
-    if group_ids:
-        principal_conditions.append(
-            and_(
-                EntityPermission.principal_type == "group",
-                EntityPermission.principal_id.in_(group_ids),
-            )
-        )
-    if role_id:
-        principal_conditions.append(
-            and_(
-                EntityPermission.principal_type == "role",
-                EntityPermission.principal_id == role_id,
-            )
-        )
-    # Guest permission — guest users have no groups/roles
-    principal_conditions.append(
-        and_(
-            EntityPermission.principal_type == "guest",
-            EntityPermission.principal_id == user_id,
-        )
-    )
-
-    perm_q = await db.execute(
-        select(EntityPermission.entity_id, EntityPermission.permission_level)
-        .where(EntityPermission.entity_type == entity_type)
-        .where(EntityPermission.tenant_id == tenant_id)
-        .where(or_(*principal_conditions))
-        .where(
-            or_(
-                EntityPermission.expires_at.is_(None),
-                EntityPermission.expires_at > now,
-            )
-        )
-    )
-    for eid, level in perm_q:
-        if eid not in visible or _rank(level) > _rank(access_map.get(eid, "none")):
-            visible.add(eid)
-            access_map[eid] = level
-
-    return visible, access_map
-
-
-async def batch_get_effective_access(
-    db: AsyncSession,
-    tenant_id: uuid.UUID,
-    user_id: uuid.UUID,
-    entity_type: str,
-    entity_ids: list[uuid.UUID],
-) -> dict[uuid.UUID, str]:
-    """Batch resolution: get access levels for multiple entities at once.
-
-    Much more efficient than calling get_effective_access() in a loop.
-    """
-    if not entity_ids:
-        return {}
-
-    # Check system admin
-    user_q = await db.execute(
-        select(User.is_system_admin).where(User.id == user_id)
-    )
-    if user_q.scalar():
-        return {eid: "delete" for eid in entity_ids}
-
-    # Get user's groups and role
-    groups_q = await db.execute(
-        select(UserGroup.group_id)
-        .where(UserGroup.user_id == user_id)
-        .where(UserGroup.tenant_id == tenant_id)
-    )
-    group_ids = [row[0] for row in groups_q]
-
-    role_q = await db.execute(
-        select(UserTenant.role_id)
-        .where(UserTenant.user_id == user_id)
-        .where(UserTenant.tenant_id == tenant_id)
-    )
-    role_id = role_q.scalar_one_or_none()
-
-    result: dict[uuid.UUID, str] = {}
-
-    # 1. Check ownership via SQLAlchemy model (safe from SQL injection)
-    model = _get_entity_model(entity_type)
-    batch_q = select(model.id, model.owner_id).where(model.id.in_(entity_ids)).where(model.tenant_id == tenant_id)
-    if hasattr(model, 'deleted_at'):
-        batch_q = batch_q.where(model.deleted_at.is_(None))
-    owner_q = await db.execute(batch_q)
-    for eid, owner_id in owner_q:
-        if owner_id == user_id:
-            result[eid] = "owner"
-        elif owner_id is None:
-            result[eid] = "read"
-
-    # 2. Check permissions
-    now = datetime.now(UTC)
-    principal_conditions = [
-        and_(
-            EntityPermission.principal_type == "user",
-            EntityPermission.principal_id == user_id,
-        ),
-    ]
-    if group_ids:
-        principal_conditions.append(
-            and_(
-                EntityPermission.principal_type == "group",
-                EntityPermission.principal_id.in_(group_ids),
-            )
-        )
-    if role_id:
-        principal_conditions.append(
-            and_(
-                EntityPermission.principal_type == "role",
-                EntityPermission.principal_id == role_id,
-            )
-        )
-    # Guest permission — guest users have no groups/roles
-    principal_conditions.append(
-        and_(
-            EntityPermission.principal_type == "guest",
-            EntityPermission.principal_id == user_id,
-        )
-    )
-
-    perm_q = await db.execute(
-        select(EntityPermission.entity_id, EntityPermission.permission_level)
-        .where(EntityPermission.entity_type == entity_type)
-        .where(EntityPermission.entity_id.in_(entity_ids))
-        .where(EntityPermission.tenant_id == tenant_id)
-        .where(or_(*principal_conditions))
-        .where(
-            or_(
-                EntityPermission.expires_at.is_(None),
-                EntityPermission.expires_at > now,
-            )
-        )
-    )
-    for eid, level in perm_q:
-        current = result.get(eid, "none")
-        if _rank(level) > _rank(current):
-            result[eid] = level
-
-    # Fill in 'none' for entities not found
-    for eid in entity_ids:
-        if eid not in result:
-            result[eid] = "none"
-
-    return result
-
-
-async def _invalidate_user_cache(
-    redis: aioredis.Redis | None,
-    tenant_id: uuid.UUID,
-    user_id: uuid.UUID,
-    entity_type: str,
-) -> None:
-    """Invalidate the visibility cache for a user + entity type."""
-    if redis is None:
-        return
-    cache_key = f"{CACHE_PREFIX}:{user_id}:{tenant_id}:{entity_type}"
-    await redis.delete(cache_key)
-
-
-async def get_cached_visible_ids(
-    db: AsyncSession,
-    redis: aioredis.Redis,
-    tenant_id: uuid.UUID,
-    user_id: uuid.UUID,
-    entity_type: str,
-) -> tuple[set[uuid.UUID], dict[uuid.UUID, str]]:
-    """Get visible IDs from Redis cache or resolve from DB.
-
-    Cache key: ep_vis:{user_id}:{tenant_id}:{entity_type}
-    Cache value: JSON {visible_ids: [str], access_map: {str: str}}
-    TTL: 5 minutes
-    """
-    cache_key = f"{CACHE_PREFIX}:{user_id}:{tenant_id}:{entity_type}"
-
-    raw = await redis.get(cache_key)
-    if raw is not None:
-        data = json.loads(raw)
-        visible = {uuid.UUID(eid) for eid in data.get("visible_ids", [])}
-        access_map = {uuid.UUID(eid): level for eid, level in data.get("access_map", {}).items()}
-        return visible, access_map
-
-    # Cache miss — resolve from DB
-    visible, access_map = await get_visible_ids(db, tenant_id, user_id, entity_type)
-
-    # Store in cache
-    cache_data = {
-        "visible_ids": [str(eid) for eid in visible],
-        "access_map": {str(eid): level for eid, level in access_map.items()},
-    }
-    await redis.setex(cache_key, CACHE_TTL, json.dumps(cache_data))
-
-    return visible, access_map
-
-
-async def invalidate_all_user_entity_cache(
-    redis: aioredis.Redis,
-    tenant_id: uuid.UUID,
-    user_id: uuid.UUID,
-) -> None:
-    """Invalidate all entity permission caches for a user."""
-    pattern = f"{CACHE_PREFIX}:{user_id}:{tenant_id}:*"
-    batch_size = 200
-    cursor: int | bytes | str = 0
-    while True:
-        cursor, keys = await redis.scan(cursor=cursor, match=pattern, count=batch_size)
-        if keys:
-            await redis.delete(*keys)
-        if int(cursor) == 0:
-            break
-
-
-async def check_entity_access(
-    db: AsyncSession,
-    tenant_id: uuid.UUID,
-    user_id: uuid.UUID,
-    entity_type: str,
-    entity_id: uuid.UUID,
-    required_level: str = "read",
-) -> bool:
-    """Check if user has at least the required access level on an entity."""
-    access = await get_effective_access(db, tenant_id, user_id, entity_type, entity_id)
-    return _rank(access) >= _rank(required_level)
-
-
 async def list_all_permissions(
     db: AsyncSession, tenant_id: uuid.UUID
 ) -> list[dict]:
@@ -930,3 +518,16 @@ async def cleanup_expired_permissions(db: AsyncSession) -> int:
         await db.commit()
         logger.info("Cleaned up %d expired entity permissions", count)
     return count
+
+
+# Backward compatibility re-exports
+from app.services.permission_resolver import (  # noqa: E402
+    get_effective_access,
+    get_visible_ids,
+    batch_get_effective_access,
+    check_entity_access,
+)
+from app.services.permission_cache import (  # noqa: E402
+    get_cached_visible_ids,
+    invalidate_all_user_entity_cache,
+)
diff --git a/app/services/permission_audit.py b/app/services/permission_audit.py
new file mode 100644
index 0000000..75e1628
--- /dev/null
+++ b/app/services/permission_audit.py
@@ -0,0 +1,97 @@
+"""Audit logging helpers for entity permission changes.
+
+Extracted from entity_permission_service.py for modularity.
+"""
+
+from __future__ import annotations
+
+import uuid
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.audit import log_audit
+from app.core.notifications import create_notification
+from app.models.entity_permission import EntityPermission
+
+
+async def _log_permission_grant(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    created_by: uuid.UUID | None,
+    perm: EntityPermission,
+    permission_level: str,
+    principal_type: str,
+    principal_id: str,
+) -> None:
+    """Log audit entry for a new permission grant and notify the user."""
+    await log_audit(
+        db, tenant_id, created_by,
+        action='permission_grant',
+        entity_type='entity_permission',
+        entity_id=perm.id,
+        changes={'permission_level': permission_level, 'principal_type': principal_type, 'principal_id': principal_id}
+    )
+    # Notify user if direct permission
+    if principal_type == 'user':
+        await create_notification(
+            db, tenant_id, uuid.UUID(principal_id),
+            type='permission_granted',
+            title='Neue Berechtigung',
+            body=f'{perm.entity_type} wurde mit dir geteilt',
+            entity_type=perm.entity_type,
+            entity_id=perm.entity_id,
+        )
+
+
+async def _log_permission_update(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    created_by: uuid.UUID | None,
+    perm: EntityPermission,
+    permission_level: str,
+    principal_type: str,
+    principal_id: str,
+) -> None:
+    """Log audit entry for a permission update and notify the user."""
+    await log_audit(
+        db, tenant_id, created_by,
+        action='permission_update',
+        entity_type='entity_permission',
+        entity_id=perm.id,
+        changes={'permission_level': permission_level, 'principal_type': principal_type, 'principal_id': principal_id}
+    )
+    # Notify user if direct permission
+    if principal_type == 'user':
+        await create_notification(
+            db, tenant_id, uuid.UUID(principal_id),
+            type='permission_granted',
+            title='Neue Berechtigung',
+            body=f'{perm.entity_type} wurde mit dir geteilt',
+            entity_type=perm.entity_type,
+            entity_id=perm.entity_id,
+        )
+
+
+async def _log_permission_revoke(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    perm: EntityPermission,
+) -> None:
+    """Log audit entry for a permission revoke and notify the user."""
+    await log_audit(
+        db, tenant_id, None,
+        action='permission_revoke',
+        entity_type='entity_permission',
+        entity_id=perm.id,
+        changes={'permission_level': perm.permission_level, 'principal_type': perm.principal_type, 'principal_id': str(perm.principal_id)}
+    )
+    # Notify user if direct permission
+    if perm.principal_type == 'user':
+        await create_notification(
+            db, tenant_id, perm.principal_id,
+            type='permission_revoked',
+            title='Berechtigung entfernt',
+            body=f'{perm.entity_type} wurde nicht mehr mit dir geteilt',
+            entity_type=perm.entity_type,
+            entity_id=perm.entity_id,
+        )
diff --git a/app/services/permission_cache.py b/app/services/permission_cache.py
new file mode 100644
index 0000000..46b15ee
--- /dev/null
+++ b/app/services/permission_cache.py
@@ -0,0 +1,82 @@
+"""Redis caching for entity permission visibility.
+
+Extracted from entity_permission_service.py for modularity.
+"""
+
+from __future__ import annotations
+
+import json
+import uuid
+
+import redis.asyncio as aioredis
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.services.permission_resolver import get_visible_ids
+
+CACHE_TTL = 300  # 5 minutes
+CACHE_PREFIX = "ep_vis"  # entity permission visibility
+
+
+async def _invalidate_user_cache(
+    redis: aioredis.Redis | None,
+    tenant_id: uuid.UUID,
+    user_id: uuid.UUID,
+    entity_type: str,
+) -> None:
+    """Invalidate the visibility cache for a user + entity type."""
+    if redis is None:
+        return
+    cache_key = f"{CACHE_PREFIX}:{user_id}:{tenant_id}:{entity_type}"
+    await redis.delete(cache_key)
+
+
+async def get_cached_visible_ids(
+    db: AsyncSession,
+    redis: aioredis.Redis,
+    tenant_id: uuid.UUID,
+    user_id: uuid.UUID,
+    entity_type: str,
+) -> tuple[set[uuid.UUID], dict[uuid.UUID, str]]:
+    """Get visible IDs from Redis cache or resolve from DB.
+
+    Cache key: ep_vis:{user_id}:{tenant_id}:{entity_type}
+    Cache value: JSON {visible_ids: [str], access_map: {str: str}}
+    TTL: 5 minutes
+    """
+    cache_key = f"{CACHE_PREFIX}:{user_id}:{tenant_id}:{entity_type}"
+
+    raw = await redis.get(cache_key)
+    if raw is not None:
+        data = json.loads(raw)
+        visible = {uuid.UUID(eid) for eid in data.get("visible_ids", [])}
+        access_map = {uuid.UUID(eid): level for eid, level in data.get("access_map", {}).items()}
+        return visible, access_map
+
+    # Cache miss — resolve from DB
+    visible, access_map = await get_visible_ids(db, tenant_id, user_id, entity_type)
+
+    # Store in cache
+    cache_data = {
+        "visible_ids": [str(eid) for eid in visible],
+        "access_map": {str(eid): level for eid, level in access_map.items()},
+    }
+    await redis.setex(cache_key, CACHE_TTL, json.dumps(cache_data))
+
+    return visible, access_map
+
+
+async def invalidate_all_user_entity_cache(
+    redis: aioredis.Redis,
+    tenant_id: uuid.UUID,
+    user_id: uuid.UUID,
+) -> None:
+    """Invalidate all entity permission caches for a user."""
+    pattern = f"{CACHE_PREFIX}:{user_id}:{tenant_id}:*"
+    batch_size = 200
+    cursor: int | bytes | str = 0
+    while True:
+        cursor, keys = await redis.scan(cursor=cursor, match=pattern, count=batch_size)
+        if keys:
+            await redis.delete(*keys)
+        if int(cursor) == 0:
+            break
diff --git a/app/services/permission_resolver.py b/app/services/permission_resolver.py
new file mode 100644
index 0000000..3ea91fa
--- /dev/null
+++ b/app/services/permission_resolver.py
@@ -0,0 +1,384 @@
+"""Permission resolution logic — effective access, visibility, batch resolution.
+
+Extracted from entity_permission_service.py for modularity.
+"""
+
+from __future__ import annotations
+
+import uuid
+from datetime import datetime, UTC
+
+from sqlalchemy import and_, or_, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.permissions import PERM_RANK as _PERM_RANK
+from app.models.entity_permission import EntityPermission
+from app.models.group import UserGroup
+from app.models.user import User, UserTenant
+
+
+def _rank(level: str) -> int:
+    return _PERM_RANK.get(level, 0)
+
+
+def _get_entity_model(entity_type: str) -> type:
+    """Get SQLAlchemy model class for entity_type, or raise ValueError.
+
+    Uses lazy import to avoid circular dependency with entity_permission_service.
+    """
+    from app.services.entity_permission_service import ENTITY_MODELS
+    model = ENTITY_MODELS.get(entity_type)
+    if model is None:
+        raise ValueError(f"Unknown entity type: {entity_type}")
+    return model
+
+
+async def get_effective_access(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    user_id: uuid.UUID,
+    entity_type: str,
+    entity_id: uuid.UUID,
+) -> str:
+    """Get the effective access level for a user on a specific entity.
+
+    Resolution (highest wins):
+    1. System admin → 'delete' (full access)
+    2. Owner → 'owner' (from owner_id on the entity table)
+    3. Direct user permission
+    4. Group permission (via user_groups)
+    5. Role permission (via user_tenants.role_id)
+    6. Guest permission (principal_type='guest')
+    7. owner_id IS NULL → 'read' (tenant-owned, visible to all with module permission)
+    8. No access → 'none'
+
+    Returns: 'none' | 'read' | 'write' | 'admin' | 'delete' | 'owner'
+    """
+    # Check system admin
+    user_q = await db.execute(
+        select(User.is_system_admin).where(User.id == user_id)
+    )
+    if user_q.scalar():
+        return "delete"
+
+    # Check ownership — load the entity's owner_id via SQLAlchemy model (safe from SQL injection)
+    model = _get_entity_model(entity_type)
+    owner_q = await db.execute(
+        select(model.owner_id).where(model.id == entity_id).where(model.tenant_id == tenant_id)
+    )
+    owner_row = owner_q.first()
+    if not owner_row:
+        return "none"
+
+    owner_id = owner_row[0]
+    if owner_id == user_id:
+        return "owner"
+
+    # Get user's groups and role
+    groups_q = await db.execute(
+        select(UserGroup.group_id)
+        .where(UserGroup.user_id == user_id)
+        .where(UserGroup.tenant_id == tenant_id)
+    )
+    group_ids = [row[0] for row in groups_q]
+
+    role_q = await db.execute(
+        select(UserTenant.role_id)
+        .where(UserTenant.user_id == user_id)
+        .where(UserTenant.tenant_id == tenant_id)
+    )
+    role_id = role_q.scalar_one_or_none()
+
+    # Build principal conditions
+    principal_conditions = [
+        and_(
+            EntityPermission.principal_type == "user",
+            EntityPermission.principal_id == user_id,
+        ),
+    ]
+    if group_ids:
+        principal_conditions.append(
+            and_(
+                EntityPermission.principal_type == "group",
+                EntityPermission.principal_id.in_(group_ids),
+            )
+        )
+    if role_id:
+        principal_conditions.append(
+            and_(
+                EntityPermission.principal_type == "role",
+                EntityPermission.principal_id == role_id,
+            )
+        )
+    # Guest permission — guests are now regular users with role=guest
+    # (Guest users may have no groups/roles, only direct entity_permissions)
+    principal_conditions.append(
+        and_(
+            EntityPermission.principal_type == "guest",
+            EntityPermission.principal_id == user_id,
+        )
+    )
+
+    # Query permissions
+    now = datetime.now(UTC)
+    perm_q = await db.execute(
+        select(EntityPermission.permission_level)
+        .where(EntityPermission.entity_type == entity_type)
+        .where(EntityPermission.entity_id == entity_id)
+        .where(EntityPermission.tenant_id == tenant_id)
+        .where(or_(*principal_conditions))
+        .where(
+            or_(
+                EntityPermission.expires_at.is_(None),
+                EntityPermission.expires_at > now,
+            )
+        )
+    )
+
+    best_level = "none"
+    for (level,) in perm_q:
+        if _rank(level) > _rank(best_level):
+            best_level = level
+
+    # If owner_id is NULL (tenant-owned), user with module permission gets at least 'read'
+    if best_level == "none" and owner_id is None:
+        return "read"
+
+    return best_level
+
+
+async def get_visible_ids(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    user_id: uuid.UUID,
+    entity_type: str,
+) -> tuple[set[uuid.UUID], dict[uuid.UUID, str]]:
+    """Get all visible entity IDs for a user and their access levels.
+
+    Returns (visible_ids, access_map) where access_map is
+    entity_id → access_level string.
+
+    Resolution:
+    1. System admin → all entities at 'delete' level
+    2. Owned entities → 'owner'
+    3. Entities with direct/group/role permissions → permission level
+    4. Tenant-owned entities (owner_id IS NULL) → 'read'
+    """
+    # Check system admin
+    user_q = await db.execute(
+        select(User.is_system_admin).where(User.id == user_id)
+    )
+    if user_q.scalar():
+        # Return all entity IDs
+        model = _get_entity_model(entity_type)
+        admin_q = select(model.id).where(model.tenant_id == tenant_id)
+        if hasattr(model, 'deleted_at'):
+            admin_q = admin_q.where(model.deleted_at.is_(None))
+        all_q = await db.execute(admin_q)
+        all_ids = {row[0] for row in all_q}
+        return all_ids, {eid: "delete" for eid in all_ids}
+
+    # Get user's groups and role
+    groups_q = await db.execute(
+        select(UserGroup.group_id)
+        .where(UserGroup.user_id == user_id)
+        .where(UserGroup.tenant_id == tenant_id)
+    )
+    group_ids = [row[0] for row in groups_q]
+
+    role_q = await db.execute(
+        select(UserTenant.role_id)
+        .where(UserTenant.user_id == user_id)
+        .where(UserTenant.tenant_id == tenant_id)
+    )
+    role_id = role_q.scalar_one_or_none()
+
+    # 1. Owned entities
+    model = _get_entity_model(entity_type)
+    owned_q_builder = select(model.id).where(model.tenant_id == tenant_id).where(model.owner_id == user_id)
+    if hasattr(model, 'deleted_at'):
+        owned_q_builder = owned_q_builder.where(model.deleted_at.is_(None))
+    owned_q = await db.execute(owned_q_builder)
+    visible: set[uuid.UUID] = set()
+    access_map: dict[uuid.UUID, str] = {}
+    for (eid,) in owned_q:
+        visible.add(eid)
+        access_map[eid] = "owner"
+
+    # 2. Tenant-owned entities (owner_id IS NULL)
+    tenant_q_builder = select(model.id).where(model.tenant_id == tenant_id).where(model.owner_id.is_(None))
+    if hasattr(model, 'deleted_at'):
+        tenant_q_builder = tenant_q_builder.where(model.deleted_at.is_(None))
+    tenant_owned_q = await db.execute(tenant_q_builder)
+    for (eid,) in tenant_owned_q:
+        if eid not in visible:
+            visible.add(eid)
+            access_map[eid] = "read"
+
+    # 3. Permission-based access
+    now = datetime.now(UTC)
+    principal_conditions = [
+        and_(
+            EntityPermission.principal_type == "user",
+            EntityPermission.principal_id == user_id,
+        ),
+    ]
+    if group_ids:
+        principal_conditions.append(
+            and_(
+                EntityPermission.principal_type == "group",
+                EntityPermission.principal_id.in_(group_ids),
+            )
+        )
+    if role_id:
+        principal_conditions.append(
+            and_(
+                EntityPermission.principal_type == "role",
+                EntityPermission.principal_id == role_id,
+            )
+        )
+    # Guest permission — guest users have no groups/roles
+    principal_conditions.append(
+        and_(
+            EntityPermission.principal_type == "guest",
+            EntityPermission.principal_id == user_id,
+        )
+    )
+
+    perm_q = await db.execute(
+        select(EntityPermission.entity_id, EntityPermission.permission_level)
+        .where(EntityPermission.entity_type == entity_type)
+        .where(EntityPermission.tenant_id == tenant_id)
+        .where(or_(*principal_conditions))
+        .where(
+            or_(
+                EntityPermission.expires_at.is_(None),
+                EntityPermission.expires_at > now,
+            )
+        )
+    )
+    for eid, level in perm_q:
+        if eid not in visible or _rank(level) > _rank(access_map.get(eid, "none")):
+            visible.add(eid)
+            access_map[eid] = level
+
+    return visible, access_map
+
+
+async def batch_get_effective_access(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    user_id: uuid.UUID,
+    entity_type: str,
+    entity_ids: list[uuid.UUID],
+) -> dict[uuid.UUID, str]:
+    """Batch resolution: get access levels for multiple entities at once.
+
+    Much more efficient than calling get_effective_access() in a loop.
+    """
+    if not entity_ids:
+        return {}
+
+    # Check system admin
+    user_q = await db.execute(
+        select(User.is_system_admin).where(User.id == user_id)
+    )
+    if user_q.scalar():
+        return {eid: "delete" for eid in entity_ids}
+
+    # Get user's groups and role
+    groups_q = await db.execute(
+        select(UserGroup.group_id)
+        .where(UserGroup.user_id == user_id)
+        .where(UserGroup.tenant_id == tenant_id)
+    )
+    group_ids = [row[0] for row in groups_q]
+
+    role_q = await db.execute(
+        select(UserTenant.role_id)
+        .where(UserTenant.user_id == user_id)
+        .where(UserTenant.tenant_id == tenant_id)
+    )
+    role_id = role_q.scalar_one_or_none()
+
+    result: dict[uuid.UUID, str] = {}
+
+    # 1. Check ownership via SQLAlchemy model (safe from SQL injection)
+    model = _get_entity_model(entity_type)
+    batch_q = select(model.id, model.owner_id).where(model.id.in_(entity_ids)).where(model.tenant_id == tenant_id)
+    if hasattr(model, 'deleted_at'):
+        batch_q = batch_q.where(model.deleted_at.is_(None))
+    owner_q = await db.execute(batch_q)
+    for eid, owner_id in owner_q:
+        if owner_id == user_id:
+            result[eid] = "owner"
+        elif owner_id is None:
+            result[eid] = "read"
+
+    # 2. Check permissions
+    now = datetime.now(UTC)
+    principal_conditions = [
+        and_(
+            EntityPermission.principal_type == "user",
+            EntityPermission.principal_id == user_id,
+        ),
+    ]
+    if group_ids:
+        principal_conditions.append(
+            and_(
+                EntityPermission.principal_type == "group",
+                EntityPermission.principal_id.in_(group_ids),
+            )
+        )
+    if role_id:
+        principal_conditions.append(
+            and_(
+                EntityPermission.principal_type == "role",
+                EntityPermission.principal_id == role_id,
+            )
+        )
+    # Guest permission — guest users have no groups/roles
+    principal_conditions.append(
+        and_(
+            EntityPermission.principal_type == "guest",
+            EntityPermission.principal_id == user_id,
+        )
+    )
+
+    perm_q = await db.execute(
+        select(EntityPermission.entity_id, EntityPermission.permission_level)
+        .where(EntityPermission.entity_type == entity_type)
+        .where(EntityPermission.entity_id.in_(entity_ids))
+        .where(EntityPermission.tenant_id == tenant_id)
+        .where(or_(*principal_conditions))
+        .where(
+            or_(
+                EntityPermission.expires_at.is_(None),
+                EntityPermission.expires_at > now,
+            )
+        )
+    )
+    for eid, level in perm_q:
+        current = result.get(eid, "none")
+        if _rank(level) > _rank(current):
+            result[eid] = level
+
+    # Fill in 'none' for entities not found
+    for eid in entity_ids:
+        if eid not in result:
+            result[eid] = "none"
+
+    return result
+
+
+async def check_entity_access(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    user_id: uuid.UUID,
+    entity_type: str,
+    entity_id: uuid.UUID,
+    required_level: str = "read",
+) -> bool:
+    """Check if user has at least the required access level on an entity."""
+    access = await get_effective_access(db, tenant_id, user_id, entity_type, entity_id)
+    return _rank(access) >= _rank(required_level)
diff --git a/app/services/user_service.py b/app/services/user_service.py
index cabcff0..7a45391 100644
--- a/app/services/user_service.py
+++ b/app/services/user_service.py
@@ -151,6 +151,7 @@ class UserService:
         email: str | None = None,
         current_password: str | None = None,
         new_password: str | None = None,
+        is_system_admin: bool | None = None,
     ) -> tuple[User, UserTenant] | None:
         """Update a user and their tenant membership.
 
@@ -181,6 +182,8 @@ class UserService:
             user_tenant.role_id = role_id
         if is_active is not None:
             user.is_active = is_active
+        if is_system_admin is not None:
+            user.is_system_admin = is_system_admin
         if first_name is not None:
             user.first_name = first_name
         if last_name is not None:
diff --git a/docker-compose.yaml b/docker-compose.yaml
index 67fdd60..4c19459 100644
--- a/docker-compose.yaml
+++ b/docker-compose.yaml
@@ -71,7 +71,7 @@ services:
       SERVICE_FQDN_CRM_APP_8000: ${APP_DOMAIN}
       DATABASE_URL: ${DATABASE_URL:-postgresql+asyncpg://crm_api:${DB_PASSWORD}@postgres:5432/${POSTGRES_DB:-crm_db}}
       AUTH_DATABASE_URL: ${AUTH_DATABASE_URL:-postgresql+asyncpg://crm_auth:${DB_PASSWORD}@postgres:5432/${POSTGRES_DB:-crm_db}}
-      MIGRATION_DATABASE_URL: ${MIGRATION_DATABASE_URL:-postgresql+asyncpg://crm_user:${DB_PASSWORD}@postgres:5432/${POSTGRES_DB:-crm_db}}
+      MIGRATION_DATABASE_URL: ${MIGRATION_DATABASE_URL:-postgresql+asyncpg://crm_migration:${DB_PASSWORD}@postgres:5432/${POSTGRES_DB:-crm_db}}
       REDIS_URL: ${REDIS_URL:-redis://default:${REDIS_PASSWORD}@redis:6379/0}
       SECRET_KEY: ${SECRET_KEY:?SECRET_KEY is required}
       CORS_ORIGINS: ${CORS_ORIGINS}
@@ -114,7 +114,7 @@ services:
     environment:
       DATABASE_URL: ${WORKER_DATABASE_URL:-postgresql+asyncpg://crm_worker:${DB_PASSWORD}@postgres:5432/${POSTGRES_DB:-crm_db}}
       WORKER_DATABASE_URL: ${WORKER_DATABASE_URL:-postgresql+asyncpg://crm_worker:${DB_PASSWORD}@postgres:5432/${POSTGRES_DB:-crm_db}}
-      MIGRATION_DATABASE_URL: ${MIGRATION_DATABASE_URL:-postgresql+asyncpg://crm_user:${DB_PASSWORD}@postgres:5432/${POSTGRES_DB:-crm_db}}
+      MIGRATION_DATABASE_URL: ${MIGRATION_DATABASE_URL:-postgresql+asyncpg://crm_migration:${DB_PASSWORD}@postgres:5432/${POSTGRES_DB:-crm_db}}
       REDIS_URL: ${REDIS_URL:-redis://default:${REDIS_PASSWORD}@redis:6379/0}
       SECRET_KEY: ${SECRET_KEY:?SECRET_KEY is required}
       FRONTEND_URL: ${FRONTEND_URL}
diff --git a/prestart.sh b/prestart.sh
index 1e993fa..b1cf61e 100644
--- a/prestart.sh
+++ b/prestart.sh
@@ -10,7 +10,7 @@
 # Notes:
 #   - `set -e` ensures the container crashes loudly if migrations fail.
 #   - The ARQ worker runs in a separate container (see worker.sh / docker-compose).
-#   - Migrations use MIGRATION_DATABASE_URL (crm_user, owner, can bypass RLS).
+#   - Migrations use MIGRATION_DATABASE_URL (crm_migration, NOSUPERUSER, BYPASSRLS).
 #   - The app uses DATABASE_URL (crm_runtime, NOSUPERUSER, NOBYPASSRLS).
 # =============================================================================