chore(cleanup): toten AI-Copilot-Legacy entfernt — Router nie gemountet, Tabellen von Migration 0137 gedroppt (Chat läuft seitdem über kommunikation/comm_conversations); schemas/OpenAPI-Tag bereinigt; Geister-Test test_ai_copilot.py geloescht (pytest.skip seit Phase 2); test_contacts_lifecycle Route-Anzahl-Failure als Vorbestand bewiesen (83 Routen auch auf clean HEAD)

app/main.py

@@ -438,7 +438,6 @@ def create_app() -> FastAPI:
             {"name": "entity-history", "description": "Audit trail and entity change history."},
             {"name": "import-export", "description": "Bulk import and export of contacts and data."},
             {"name": "plugins", "description": "Plugin management: list, install, activate, deactivate."},
-            {"name": "ai-copilot", "description": "AI copilot: chat, suggestions, conversation history."},
             {"name": "workflows", "description": "Workflow definitions, instances, and execution."},
             {"name": "user-preferences", "description": "Per-user preference settings."},
             {"name": "currencies", "description": "Currency management for multi-currency support."},

app/models/ai_conversation.py

@@ -1,53 +0,0 @@
-"""AI Conversation and Message models — tenant-scoped with RLS."""
-
-from __future__ import annotations
-
-import uuid
-from typing import Any
-
-from sqlalchemy import ForeignKey, Index, Integer, String, Text
-from sqlalchemy.dialects.postgresql import JSONB
-from sqlalchemy.dialects.postgresql import UUID as PGUUID
-from sqlalchemy.orm import Mapped, mapped_column
-
-from app.core.db import Base, TenantMixin
-from app.models.owned_mixin import OwnedMixin
-
-
-class AIConversation(Base, TenantMixin, OwnedMixin):
-    """AI Copilot conversation thread — tenant-scoped."""
-
-    __tablename__ = "ai_conversations"
-    __table_args__ = (Index("ix_ai_conversations_tenant_user", "tenant_id", "user_id"),)
-
-    id: Mapped[uuid.UUID] = mapped_column(
-        PGUUID(as_uuid=True), primary_key=True, default=uuid.uuid4
-    )
-    user_id: Mapped[uuid.UUID] = mapped_column(
-        PGUUID(as_uuid=True), ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True
-    )
-    title: Mapped[str] = mapped_column(String(255), nullable=False, default="Untitled")
-    context: Mapped[dict[str, Any]] = mapped_column(JSONB, default=dict, nullable=False)
-
-
-class AIMessage(Base, TenantMixin, OwnedMixin):
-    """Individual messages within an AI conversation — user input, AI response, actions."""
-
-    __tablename__ = "ai_messages"
-    __table_args__ = (Index("ix_ai_messages_tenant_conversation", "tenant_id", "conversation_id"),)
-
-    id: Mapped[uuid.UUID] = mapped_column(
-        PGUUID(as_uuid=True), primary_key=True, default=uuid.uuid4
-    )
-    conversation_id: Mapped[uuid.UUID] = mapped_column(
-        PGUUID(as_uuid=True),
-        ForeignKey("ai_conversations.id", ondelete="CASCADE"),
-        nullable=False,
-        index=True,
-    )
-    role: Mapped[str] = mapped_column(String(20), nullable=False)  # user, assistant, system
-    content: Mapped[str] = mapped_column(Text, nullable=False)
-    proposed_actions: Mapped[list[dict[str, Any]] | None] = mapped_column(JSONB, nullable=True)
-    executed_action: Mapped[dict[str, Any] | None] = mapped_column(JSONB, nullable=True)
-    execution_result: Mapped[dict[str, Any] | None] = mapped_column(JSONB, nullable=True)
-    message_index: Mapped[int] = mapped_column(Integer, nullable=False, default=0)

app/routes/ai_copilot.py

@@ -1,135 +0,0 @@
-"""AI Copilot routes — query, execute, history.
-
-.. deprecated:: Phase 2 (ARCH-059)
-    Legacy module, not wired into main.py and without frontend consumers.
-    Scheduled for removal together with app/services/ai_copilot_service.py
-    and the ``ai_conversations`` tables (see service docstring for plan).
-"""
-
-from __future__ import annotations
-
-import uuid
-
-from fastapi import APIRouter, Depends, HTTPException, Query, Request, status
-from sqlalchemy.ext.asyncio import AsyncSession
-
-from app.core.db import get_db
-from app.core.rate_limit import RateLimitPolicy, check_rate_limit_policy
-from app.deps import require_permission
-from app.schemas.ai_copilot import (
-    CopilotExecuteRequest,
-    CopilotQueryRequest,
-)
-from app.services import ai_copilot_service
-
-router = APIRouter(prefix="/api/v1/ai/copilot", tags=["ai-copilot"])
-
-
-@router.post("/query")
-async def copilot_query(
-    body: CopilotQueryRequest,
-    request: Request,
-    db: AsyncSession = Depends(get_db),
-    current_user: dict = Depends(require_permission("ai:write")),
-):
-    """Process a natural language query and return proposed actions.
-
-    Returns proposed_actions array for user confirmation.
-    Does NOT execute any actions — user must call /execute to confirm.
-    """
-    tenant_id = uuid.UUID(current_user["tenant_id"])
-    user_id = uuid.UUID(current_user["user_id"])
-
-    # Rate limit — AI policy (cost-sensitive LLM call)
-    await check_rate_limit_policy(
-        f"rate:ai:copilot:{tenant_id}:{user_id}",
-        RateLimitPolicy.AI,
-    )
-
-    result = await ai_copilot_service.process_query(
-        db,
-        tenant_id,
-        user_id,
-        query=body.query,
-        conversation_id=body.conversation_id,
-        context=body.context,
-    )
-
-    if "error" in result and result.get("status_code") == 404:
-        raise HTTPException(
-            status_code=status.HTTP_404_NOT_FOUND,
-            detail={"detail": result["error"], "code": "conversation_not_found"},
-        )
-
-    return result
-
-
-@router.post("/execute")
-async def copilot_execute(
-    body: CopilotExecuteRequest,
-    request: Request,
-    db: AsyncSession = Depends(get_db),
-    current_user: dict = Depends(require_permission("ai:write")),
-):
-    """Execute a proposed action after user confirmation.
-
-    RBAC is enforced — the user must have permission for the action.
-    Returns 200 with execution result or 403 if RBAC blocks.
-    """
-    tenant_id = uuid.UUID(current_user["tenant_id"])
-    user_id = uuid.UUID(current_user["user_id"])
-
-    # Rate limit — AI policy (cost-sensitive LLM call)
-    await check_rate_limit_policy(
-        f"rate:ai:copilot:{tenant_id}:{user_id}",
-        RateLimitPolicy.AI,
-    )
-    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,
-        resolved,
-        conversation_id=body.conversation_id,
-        action=body.action.model_dump(),
-    )
-
-    if result.get("status_code") == 404:
-        raise HTTPException(
-            status_code=status.HTTP_404_NOT_FOUND,
-            detail={"detail": result.get("error", "Not found"), "code": "not_found"},
-        )
-
-    if result.get("status_code") == 403:
-        raise HTTPException(
-            status_code=status.HTTP_403_FORBIDDEN,
-            detail={"detail": result.get("error", "Insufficient permissions"), "code": "forbidden"},
-        )
-
-    return result
-
-
-@router.get("/history")
-async def copilot_history(
-    page: int = Query(1, ge=1),
-    page_size: int = Query(20, ge=1, le=100),
-    db: AsyncSession = Depends(get_db),
-    current_user: dict = Depends(require_permission("ai:read")),
-):
-    """Get paginated conversation history for the current user."""
-    tenant_id = uuid.UUID(current_user["tenant_id"])
-    user_id = uuid.UUID(current_user["user_id"])
-
-    return await ai_copilot_service.get_history(
-        db,
-        tenant_id,
-        user_id,
-        page=page,
-        page_size=page_size,
-    )

app/schemas/__init__.py

@@ -1,14 +1,5 @@
 """Pydantic schemas package."""
 
-from app.schemas.ai_copilot import (
-    CopilotAction,  # noqa: F401
-    CopilotExecuteRequest,  # noqa: F401
-    CopilotExecuteResponse,  # noqa: F401
-    CopilotHistoryResponse,  # noqa: F401
-    CopilotMessageResponse,  # noqa: F401
-    CopilotQueryRequest,  # noqa: F401
-    CopilotQueryResponse,  # noqa: F401
-)
 from app.schemas.plugin import (
     PluginActionResponse,  # noqa: F401
     PluginInfo,  # noqa: F401

app/schemas/ai_copilot.py

@@ -1,69 +0,0 @@
-"""AI Copilot schemas — query, execute, history."""
-
-from __future__ import annotations
-
-from pydantic import BaseModel, Field
-
-
-class CopilotQueryRequest(BaseModel):
-    """Natural language query to the AI copilot."""
-
-    query: str = Field(..., min_length=1, max_length=2000)
-    conversation_id: str | None = None
-    context: dict = Field(default_factory=dict)
-
-
-class CopilotAction(BaseModel):
-    """A proposed API action derived from NL input."""
-
-    method: str = Field(..., pattern="^(GET|POST|PATCH|DELETE)$")
-    path: str = Field(..., min_length=1)
-    body: dict | None = None
-    description: str = ""
-    confidence: float = Field(0.0, ge=0.0, le=1.0)
-
-
-class CopilotQueryResponse(BaseModel):
-    """Response from copilot query — proposed actions for user confirmation."""
-
-    conversation_id: str
-    message: str
-    proposed_actions: list[CopilotAction] = Field(default_factory=list)
-
-
-class CopilotExecuteRequest(BaseModel):
-    """Execute a proposed action after user confirmation."""
-
-    conversation_id: str
-    action: CopilotAction
-
-
-class CopilotExecuteResponse(BaseModel):
-    """Result of executing a proposed action."""
-
-    conversation_id: str
-    success: bool
-    status_code: int
-    data: dict | list | None = None
-    error: str | None = None
-
-
-class CopilotMessageResponse(BaseModel):
-    """A single message in conversation history."""
-
-    id: str
-    role: str
-    content: str
-    proposed_actions: list[dict] | None = None
-    executed_action: dict | None = None
-    execution_result: dict | None = None
-    created_at: str | None = None
-
-
-class CopilotHistoryResponse(BaseModel):
-    """Paginated conversation history."""
-
-    items: list[CopilotMessageResponse]
-    total: int
-    page: int
-    page_size: int

app/services/ai_copilot_service.py

@@ -1,477 +0,0 @@
-"""AI Copilot service — NL query processing, action execution, RBAC, audit logging.
-
-.. deprecated:: Phase 2
-    This module is a legacy leftover of the pre-kommunikation AI chat system.
-    It is NOT registered in main.py, has NO frontend consumers, and its test
-    suite is skipped ("ai_copilot routes removed in Phase 2").
-
-    Shutdown plan (ARCH-059):
-    1. ✅ Marked deprecated (this notice)
-    2. Remove routes/service/model + ``ai_conversations`` tables in a dedicated
-       migration once a release confirms zero traffic on /api/v1/ai/copilot/*
-    3. Do NOT migrate to kommunikation — the plugin already covers chat via
-       CommConversation/CommMessage; parity work would be wasted effort.
-"""
-
-from __future__ import annotations
-
-import uuid
-import warnings
-from typing import Any
-
-from sqlalchemy import desc, func, select
-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.permissions import check_permission
-from app.core.visibility import apply_visibility_filter, check_single_entity_access
-
-try:
-    from app.models.ai_conversation import AIConversation, AIMessage
-except ImportError:
-    AIConversation = None  # type: ignore
-    AIMessage = None  # type: ignore
-from app.models.contact import Contact
-from app.models.workflow import Workflow
-
-warnings.warn(
-    "app.services.ai_copilot_service is deprecated (ARCH-059) — "
-    "scheduled for removal; do not build new features on it",
-    DeprecationWarning,
-    stacklevel=2,
-)
-
-
-def _safe_iso(dt) -> str | None:
-    if dt is None:
-        return None
-    try:
-        return dt.isoformat() if hasattr(dt, "isoformat") else None
-    except Exception:
-        return None
-
-
-def _get_attr(obj, name, default=None):
-    try:
-        val = getattr(obj, name)
-        return val if val is not None else default
-    except Exception:
-        return default
-
-
-def _conversation_to_dict(c: AIConversation) -> dict[str, Any]:
-    return {
-        "id": str(c.id),
-        "title": c.title,
-        "context": c.context,
-        "created_at": _safe_iso(_get_attr(c, "created_at")),
-        "updated_at": _safe_iso(_get_attr(c, "updated_at")),
-    }
-
-
-def _message_to_dict(m: AIMessage) -> dict[str, Any]:
-    return {
-        "id": str(m.id),
-        "role": m.role,
-        "content": m.content,
-        "proposed_actions": m.proposed_actions,
-        "executed_action": m.executed_action,
-        "execution_result": m.execution_result,
-        "created_at": _safe_iso(_get_attr(m, "created_at")),
-    }
-
-
-async def process_query(
-    db: AsyncSession,
-    tenant_id: uuid.UUID,
-    user_id: uuid.UUID,
-    query: str,
-    conversation_id: str | None = None,
-    context: dict[str, Any] | None = None,
-    is_system_admin: bool = False,
-) -> dict[str, Any]:
-    """Process a natural language query and return proposed actions.
-
-    1. Get or create conversation
-    2. Store user message
-    3. Call LLM client (mock or real) to get proposed actions
-    4. Store assistant message with proposed actions
-    5. Return response with conversation_id and proposed_actions
-    """
-    context = context or {}
-
-    # Get or create conversation
-    if conversation_id:
-        conv_uuid = uuid.UUID(conversation_id)
-        result = await db.execute(
-            select(AIConversation).where(
-                AIConversation.id == conv_uuid,
-                AIConversation.tenant_id == tenant_id,
-            )
-        )
-        conversation = result.scalar_one_or_none()
-        if conversation is None:
-            return {"error": "Conversation not found", "status_code": 404}
-    else:
-        conversation = AIConversation(
-            tenant_id=tenant_id,
-            user_id=user_id,
-            title=query[:100] if query else "Untitled",
-            context=context,
-        )
-        db.add(conversation)
-        await db.flush()
-        await db.refresh(conversation)
-
-    # Get next message index
-    count_q = select(func.count()).select_from(
-        select(AIMessage).where(AIMessage.conversation_id == conversation.id).subquery()
-    )
-    count_result = await db.execute(count_q)
-    msg_index = count_result.scalar_one()
-
-    # Store user message
-    user_msg = AIMessage(
-        tenant_id=tenant_id,
-        conversation_id=conversation.id,
-        role="user",
-        content=query,
-        message_index=msg_index,
-    )
-    db.add(user_msg)
-    await db.flush()
-    await db.refresh(user_msg)
-
-    # Call LLM client
-    llm = get_llm_client()
-    llm_response = await llm.generate(query, context)
-
-    # Store assistant message
-    assistant_msg = AIMessage(
-        tenant_id=tenant_id,
-        conversation_id=conversation.id,
-        role="assistant",
-        content=llm_response.message,
-        proposed_actions=llm_response.proposed_actions,
-        message_index=msg_index + 1,
-    )
-    db.add(assistant_msg)
-    await db.flush()
-    await db.refresh(assistant_msg)
-
-    # Log to audit
-    await log_audit(
-        db,
-        tenant_id,
-        user_id,
-        action="query",
-        entity_type="ai_copilot",
-        entity_id=conversation.id,
-        changes={"query": query, "proposed_action_count": len(llm_response.proposed_actions)},
-    )
-
-    return {
-        "conversation_id": str(conversation.id),
-        "message": llm_response.message,
-        "proposed_actions": llm_response.proposed_actions,
-    }
-
-
-async def execute_action(
-    db: AsyncSession,
-    tenant_id: uuid.UUID,
-    user_id: uuid.UUID,
-    resolved: dict[str, Any],
-    conversation_id: str,
-    action: dict[str, Any],
-    is_system_admin: bool = False,
-) -> dict[str, Any]:
-    """Execute a proposed action with RBAC enforcement.
-
-    1. Validate conversation belongs to tenant
-    2. Check RBAC permissions for the action
-    3. Execute the action (direct DB or API call)
-    4. Store execution result in message
-    5. Log to audit
-    """
-    conv_uuid = uuid.UUID(conversation_id)
-
-    # Validate conversation ownership
-    result = await db.execute(
-        select(AIConversation).where(
-            AIConversation.id == conv_uuid,
-            AIConversation.tenant_id == tenant_id,
-        )
-    )
-    conversation = result.scalar_one_or_none()
-    if conversation is None:
-        return {"error": "Conversation not found", "status_code": 404}
-
-    method = action.get("method", "GET").upper()
-    path = action.get("path", "")
-    body = action.get("body") or {}
-
-    # 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(resolved, required_perm):
-        return {
-            "error": "Insufficient permissions for this action",
-            "status_code": 403,
-            "success": False,
-        }
-
-    # Check single entity access for write operations
-    if method in ("POST", "PATCH", "DELETE"):
-        parts = path.replace("/api/v1/", "").strip("/").split("/")
-        entity_type = parts[0] if parts else ""
-        entity_id = parts[1] if len(parts) > 1 else None
-        if entity_id:
-            try:
-                entity_uuid = uuid.UUID(entity_id)
-            except (ValueError, TypeError):
-                entity_uuid = None
-            if entity_uuid:
-                has_access = await check_single_entity_access(
-                    db, entity_type, entity_uuid, user_id, tenant_id,
-                    required_level="write", is_system_admin=is_system_admin,
-                )
-                if not has_access:
-                    return {
-                        "error": "Insufficient access to this entity",
-                        "status_code": 403,
-                        "success": False,
-                    }
-
-    # Execute the action
-    try:
-        exec_result = await _execute_api_action(db, tenant_id, user_id, method, path, body, is_system_admin=is_system_admin)
-    except Exception as exc:
-        exec_result = {"error": str(exc), "status_code": 500}
-
-    # Get next message index
-    count_q = select(func.count()).select_from(
-        select(AIMessage).where(AIMessage.conversation_id == conversation.id).subquery()
-    )
-    count_result = await db.execute(count_q)
-    msg_index = count_result.scalar_one()
-
-    # Store execution message
-    exec_msg = AIMessage(
-        tenant_id=tenant_id,
-        conversation_id=conversation.id,
-        role="assistant",
-        content=f"Executed {method} {path}",
-        executed_action=action,
-        execution_result=exec_result,
-        message_index=msg_index,
-    )
-    db.add(exec_msg)
-    await db.flush()
-    await db.refresh(exec_msg)
-
-    # Log to audit
-    await log_audit(
-        db,
-        tenant_id,
-        user_id,
-        action="execute",
-        entity_type="ai_copilot",
-        entity_id=conversation.id,
-        changes={"action": action, "result": exec_result},
-    )
-
-    return {
-        "conversation_id": str(conversation.id),
-        "success": exec_result.get("success", True),
-        "status_code": exec_result.get("status_code", 200),
-        "data": exec_result.get("data"),
-        "error": exec_result.get("error"),
-    }
-
-
-async def get_history(
-    db: AsyncSession,
-    tenant_id: uuid.UUID,
-    user_id: uuid.UUID,
-    page: int = 1,
-    page_size: int = 20,
-) -> dict[str, Any]:
-    """Get paginated conversation history for the current user."""
-    page = max(1, page)
-    page_size = max(1, min(100, page_size))
-
-    base = select(AIConversation).where(
-        AIConversation.tenant_id == tenant_id,
-        AIConversation.user_id == user_id,
-    )
-
-    count_q = select(func.count()).select_from(base.subquery())
-    total_result = await db.execute(count_q)
-    total = total_result.scalar_one()
-
-    offset = (page - 1) * page_size
-    paginated = base.order_by(desc(AIConversation.created_at)).offset(offset).limit(page_size)
-    result = await db.execute(paginated)
-    conversations = result.scalars().all()
-
-    items: list[dict[str, Any]] = []
-    for conv in conversations:
-        # Get messages for each conversation
-        msg_q = (
-            select(AIMessage)
-            .where(
-                AIMessage.conversation_id == conv.id,
-                AIMessage.tenant_id == tenant_id,
-            )
-            .order_by(AIMessage.message_index)
-        )
-        msg_result = await db.execute(msg_q)
-        messages = msg_result.scalars().all()
-
-        items.append(
-            {
-                **_conversation_to_dict(conv),
-                "messages": [_message_to_dict(m) for m in messages],
-            }
-        )
-
-    return {
-        "items": items,
-        "total": total,
-        "page": page,
-        "page_size": page_size,
-    }
-
-
-async def _execute_api_action(
-    db: AsyncSession,
-    tenant_id: uuid.UUID,
-    user_id: uuid.UUID,
-    method: str,
-    path: str,
-    body: dict[str, Any],
-    is_system_admin: bool = False,
-) -> dict[str, Any]:
-    """Execute an API action directly against the database.
-
-    Supports contacts CRUD (including company-type contacts), plus workflow listing.
-    """
-    # Parse path to determine entity and operation
-    parts = path.replace("/api/v1/", "").strip("/").split("/")
-    entity = parts[0] if parts else ""
-    entity_id = parts[1] if len(parts) > 1 else None
-
-    if entity in ("companies", "contacts"):
-        return await _exec_contacts(db, tenant_id, user_id, method, entity_id, body, is_system_admin=is_system_admin)
-    elif entity == "workflows":
-        return await _exec_workflows(db, tenant_id, user_id, method, entity_id, body, is_system_admin=is_system_admin)
-    else:
-        return {"error": f"Unsupported entity: {entity}", "status_code": 400, "success": False}
-
-
-async def _exec_contacts(
-    db: AsyncSession,
-    tenant_id: uuid.UUID,
-    user_id: uuid.UUID,
-    method: str,
-    entity_id: str | None,
-    body: dict[str, Any],
-    is_system_admin: bool = False,
-) -> dict[str, Any]:
-    """Execute contact operations (unified: company + person)."""
-    if method == "GET":
-        query = select(Contact).where(
-            Contact.tenant_id == tenant_id,
-            Contact.deleted_at.is_(None),
-        )
-        query = await apply_visibility_filter(
-            db, query, "contact", Contact, user_id, tenant_id, is_system_admin=is_system_admin
-        )
-        result = await db.execute(query)
-        contacts = result.scalars().all()
-        return {
-            "success": True,
-            "status_code": 200,
-            "data": [{"id": str(c.id), "name": c.name or c.displayname, "email": c.email_1, "type": c.type} for c in contacts],
-        }
-
-    elif method == "POST":
-        contact = Contact(
-            tenant_id=tenant_id,
-            type=body.get("type", "company"),
-            name=body.get("name", "Untitled"),
-            email_1=body.get("email"),
-            phone_1=body.get("phone"),
-            created_by=user_id,
-            updated_by=user_id,
-        )
-        db.add(contact)
-        await db.flush()
-        return {
-            "success": True,
-            "status_code": 201,
-            "data": {"id": str(contact.id), "name": contact.name, "type": contact.type},
-        }
-
-    return {"error": f"Unsupported method: {method}", "status_code": 400, "success": False}
-
-
-async def _exec_workflows(
-    db: AsyncSession,
-    tenant_id: uuid.UUID,
-    user_id: uuid.UUID,
-    method: str,
-    entity_id: str | None,
-    body: dict[str, Any],
-    is_system_admin: bool = False,
-) -> dict[str, Any]:
-    """Execute workflow operations."""
-    if method == "GET":
-        result = await db.execute(
-            select(Workflow).where(
-                Workflow.tenant_id == tenant_id,
-                Workflow.is_active.is_(True),
-            )
-        )
-        workflows = result.scalars().all()
-        return {
-            "success": True,
-            "status_code": 200,
-            "data": [
-                {"id": str(w.id), "name": w.name, "trigger_event": w.trigger_event}
-                for w in workflows
-            ],
-        }
-
-    return {"error": f"Unsupported method: {method}", "status_code": 400, "success": False}
-
-
-def _derive_rbac_from_path(method: str, path: str) -> tuple[str, str]:
-    """Derive module and action_type from HTTP method and path for RBAC.
-
-    Returns (module, action_type) suitable for check_permission().
-    """
-    parts = path.replace("/api/v1/", "").strip("/").split("/")
-    entity = parts[0] if parts else ""
-
-    method_to_action = {
-        "GET": "read",
-        "POST": "create",
-        "PATCH": "update",
-        "DELETE": "delete",
-    }
-    action_type = method_to_action.get(method, "read")
-
-    # Map path entities to permission modules
-    entity_to_module = {
-        "companies": "companies",
-        "contacts": "contacts",
-        "workflows": "workflows",
-        "ai": "ai_copilot",
-    }
-    module = entity_to_module.get(entity, entity)
-
-    return module, action_type