leocrm/app/ai/agent_permissions.py

"""Agent permission context resolution for AI agents.

Resolves the effective permissions available to an agent run as the
intersection of the user's (or run-as user's) RBAC permissions, the agent's
configured tools/skills, and the tools each skill is allowed to use.

Effective = User/Run-as ∩ Agent ∩ Skill ∩ Tool

Key principles:
- Skills orchestrate tools but NEVER grant additional permissions.
- Every tool/service call re-checks permissions — rights are NOT frozen
  for a run.
- System admins get all tools.
"""

from __future__ import annotations

import logging
import uuid
from dataclasses import dataclass
from typing import Any

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.error_codes import ApiError
from app.core.permissions import check_permission, resolve_permissions

logger = logging.getLogger(__name__)


@dataclass
class AgentPermissionContext:
    """Effective permission context for a single agent run."""

    user_id: uuid.UUID
    tenant_id: uuid.UUID
    run_as_user_id: uuid.UUID | None
    user_permissions: dict[str, Any]  # RBAC permissions from Role
    agent_tool_ids: list[str]
    agent_skill_ids: list[str]
    effective_tool_ids: list[str]  # After intersection
    is_system_admin: bool = False

    def has_permission(self, permission: str) -> bool:
        """Check whether the run-as user has the given RBAC permission."""
        return check_permission(self.user_permissions, permission)

    def can_use_tool(self, tool_id: str) -> bool:
        """Check whether the agent may call the given tool."""
        return tool_id in self.effective_tool_ids


def _resolve_effective_tool_ids(
    agent_definition: Any,
    user_permissions: dict[str, Any],
) -> list[str]:
    """Compute the effective tool IDs after User ∩ Agent ∩ Skill ∩ Tool.

    Mirrors the semantics of ``app.ai.agent_tools.get_agent_tools``: skills
    orchestrate tools but never grant additional permissions.
    """
    from app.ai.skill_registry import get_skill_registry
    from app.ai.tool_registry import get_tool_registry

    agent_tool_ids: list[str] = list(getattr(agent_definition, "tool_ids", None) or [])
    agent_skill_ids: list[str] = list(getattr(agent_definition, "skill_ids", None) or [])

    skill_registry = get_skill_registry()
    skills = skill_registry.get_by_names(agent_skill_ids)

    direct_tool_ids = set(agent_tool_ids)
    skill_tool_ids: set[str] = set()
    for skill in skills:
        skill_tool_ids.update(skill.allowed_tool_ids or [])

    # Tools directly on the agent, plus tools reachable via skills that are
    # also directly on the agent (skills never widen the agent's tool set).
    available_tool_ids = direct_tool_ids | (direct_tool_ids & skill_tool_ids)

    tool_registry = get_tool_registry()
    tools = tool_registry.get_by_names(sorted(available_tool_ids))

    permitted = [
        tool
        for tool in tools
        if not getattr(tool, "required_permission", None)
        or check_permission(user_permissions, tool.required_permission)
    ]
    return [tool.name for tool in permitted]


async def resolve_agent_permissions(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    user_id: uuid.UUID,
    agent_definition: Any,
    run_as_user_id: uuid.UUID | None = None,
) -> AgentPermissionContext:
    """Resolve effective permissions for an agent run.

    Effective = User/Run-as ∩ Agent ∩ Skill ∩ Tool.

    Args:
        db: Async DB session.
        tenant_id: Tenant ID for multi-tenancy.
        user_id: The user requesting the run (permission source).
        agent_definition: AgentDefinition with tool_ids and skill_ids.
        run_as_user_id: Optional user the agent runs as. When provided, the
            run-as user's permissions are used instead of the requester's.

    Returns:
        AgentPermissionContext with the resolved effective tool IDs.
    """
    effective_user_id = run_as_user_id or user_id
    user_permissions = await resolve_permissions(db, effective_user_id, tenant_id)
    is_system_admin = bool(user_permissions.get("is_system_admin", False))

    agent_tool_ids: list[str] = list(getattr(agent_definition, "tool_ids", None) or [])
    agent_skill_ids: list[str] = list(getattr(agent_definition, "skill_ids", None) or [])

    if is_system_admin:
        effective_tool_ids = list(agent_tool_ids)
    else:
        effective_tool_ids = _resolve_effective_tool_ids(agent_definition, user_permissions)

    return AgentPermissionContext(
        user_id=user_id,
        tenant_id=tenant_id,
        run_as_user_id=run_as_user_id,
        user_permissions=user_permissions,
        agent_tool_ids=agent_tool_ids,
        agent_skill_ids=agent_skill_ids,
        effective_tool_ids=effective_tool_ids,
        is_system_admin=is_system_admin,
    )


async def check_entity_lock(
    db: AsyncSession,
    entity_type: str,
    entity_id: uuid.UUID,
    expected_version: int,
) -> bool:
    """Optimistic-lock check: raise ApiError('conflict') on version mismatch.

    Loads the entity's ``version`` column. If the current version differs from
    ``expected_version``, raises ``ApiError`` with code ``conflict``. Models
    without a ``version`` column are treated as unlocked (no-op).

    Returns True when the lock check passes.
    """
    from app.services.entity_permission_service import ENTITY_MODELS

    model = ENTITY_MODELS.get(entity_type)
    if model is None or not hasattr(model, "version"):
        return True

    result = await db.execute(select(model.version).where(model.id == entity_id))
    current_version = result.scalar_one_or_none()
    if current_version is None:
        raise ApiError(code="not_found", detail=f"{entity_type} not found")

    if int(current_version) != int(expected_version):
        raise ApiError(
            code="conflict",
            detail=(
                f"{entity_type} {entity_id} was modified concurrently "
                f"(expected version {expected_version}, current {current_version})"
            ),
        )
    return True


async def filter_visible_agents(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    user_id: uuid.UUID,
    agents: list,
) -> list:
    """Filter agents visible to a user based on agents:read + EntityPermission.

    A user sees an agent when they have the ``agents:read`` permission AND the
    agent is visible via ownership, tenant-ownership, or entity_permissions.
    System admins see all agents.
    """
    user_permissions = await resolve_permissions(db, user_id, tenant_id)
    if user_permissions.get("is_system_admin"):
        return list(agents)
    if not check_permission(user_permissions, "agents:read"):
        return []

    from app.services.permission_resolver import get_visible_ids

    visible_ids, _ = await get_visible_ids(db, tenant_id, user_id, "agent_definition")
    return [a for a in agents if a.id in visible_ids]


async def check_agent_execute_permission(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    user_id: uuid.UUID,
    agent_id: uuid.UUID,
) -> bool:
    """Check if a user has agents:execute permission for a specific agent.

    Requires the ``agents:execute`` RBAC permission AND entity-level access
    (owner, tenant-owned, or shared via entity_permissions). System admins
    always pass.
    """
    user_permissions = await resolve_permissions(db, user_id, tenant_id)
    if user_permissions.get("is_system_admin"):
        return True
    if not check_permission(user_permissions, "agents:execute"):
        return False

    from app.services.permission_resolver import check_entity_access

    return await check_entity_access(
        db, tenant_id, user_id, "agent_definition", agent_id, required_level="read"
    )


__all__ = [
    "AgentPermissionContext",
    "resolve_agent_permissions",
    "check_entity_lock",
    "filter_visible_agents",
    "check_agent_execute_permission",
]