feat(admin): пользователи, роли, матрица прав — API с аудитом и защитой LAST_ADMIN

/api/v1/admin/* под правом USERS_ADMIN (403 на всё без него): список пользователей
(поиск по email/ФИО, is_active, Page), создание (email уникален без учёта регистра →
409 EMAIL_EXISTS, роли — коды активных ролей → 422 UNKNOWN_ROLE), PATCH полей и
деактивация, PUT ролей (замена набора с diff), GET ролей с правами, GET справочника
прав (16 кодов), PUT прав роли (422 UNKNOWN_PERMISSION, granted_by/granted_at).
Аудит USER_CHANGE / USER_ROLE_CHANGE (USERS) и ROLE_PERMISSION_CHANGE (ACCESS);
повтор того же состояния аудит не пишет.

LAST_ADMIN: деактивация (в том числе себя), замена ролей и снятие USERS_ADMIN с роли
запрещены, если у активных пользователей через активные роли не останется USERS_ADMIN.
Гонки: строка пользователя/роли и строка права USERS_ADMIN блокируются FOR NO KEY
UPDATE (FOR UPDATE давал deadlock на FK-проверке вставки audit_event); тест двух
параллельных транзакций с барьером.

Задача плана Task 12, tasks.md 4.1.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2

api/src/api/dependencies/admin.py

@@ -0,0 +1,23 @@
+"""Зависимости API администрирования (spec users-admin): право `USERS_ADMIN` и сервис (R-6: сессия — `SessionDep`)."""
+
+from typing import Annotated
+
+from fastapi import Depends
+
+from src.api.dependencies.audit import RequestAudit, get_request_audit
+from src.api.dependencies.auth import require
+from src.api.dependencies.db import SessionDep
+from src.services.users_admin import USERS_ADMIN, UsersAdminService
+
+# Право на все маршруты `/admin/*`: без него — 403 до любых проверок тела и существования объектов
+ADMIN_ONLY = [Depends(require(USERS_ADMIN))]
+
+
+def get_users_admin_service(
+    session: SessionDep,
+    audit: Annotated[RequestAudit, Depends(get_request_audit)],
+) -> UsersAdminService:
+    return UsersAdminService(session, audit, audit.user_id)
+
+
+UsersAdminServiceDep = Annotated[UsersAdminService, Depends(get_users_admin_service)]

api/src/api/endpoints/admin/__init__.py

@@ -0,0 +1,10 @@
+"""Маршруты администрирования `/api/v1/admin/*` (spec users-admin): все требуют право `USERS_ADMIN`."""
+
+from fastapi import APIRouter
+
+from src.api.dependencies.admin import ADMIN_ONLY
+from src.api.endpoints.admin import roles, users
+
+router = APIRouter(prefix="/admin", tags=["admin"], dependencies=ADMIN_ONLY)
+router.include_router(users.router)
+router.include_router(roles.router)

api/src/api/endpoints/admin/roles.py

@@ -0,0 +1,42 @@
+"""Матрица прав `/admin/roles`, `/admin/permissions` (spec users-admin): чтение и замена набора прав роли.
+
+Права — справочник только для чтения (новые коды не создаются); роли не создаются, не переименовываются и не
+деактивируются через API (системные роли — `is_system`, D-8).
+"""
+
+from fastapi import APIRouter
+
+from src.api.dependencies.admin import UsersAdminServiceDep
+from src.models.access import Role
+from src.schemas.admin import PermissionOut, PermissionsSet, RoleOut
+
+router = APIRouter()
+
+
+def _out(role: Role, permissions: list[str]) -> RoleOut:
+    return RoleOut(
+        code=role.code,
+        name=role.name,
+        description=role.description,
+        is_system=role.is_system,
+        is_active=role.is_active,
+        permissions=permissions,
+    )
+
+
+@router.get("/roles", response_model=list[RoleOut], summary="Роли с правами")
+async def list_roles(service: UsersAdminServiceDep) -> list[RoleOut]:
+    return [_out(role, permissions) for role, permissions in await service.list_roles()]
+
+
+@router.get("/permissions", response_model=list[PermissionOut], summary="Справочник прав")
+async def list_permissions(service: UsersAdminServiceDep) -> list[PermissionOut]:
+    return [PermissionOut.model_validate(permission) for permission in await service.list_permissions()]
+
+
+@router.put("/roles/{code}/permissions", response_model=RoleOut, summary="Заменить права роли")
+async def set_role_permissions(code: str, data: PermissionsSet, service: UsersAdminServiceDep) -> RoleOut:
+    """Неизвестная роль — 404; неизвестный код права — 422 `UNKNOWN_PERMISSION`; снятие последнего `USERS_ADMIN` —
+    409 `LAST_ADMIN`."""
+    await service.set_role_permissions(code, data.permissions)
+    return _out(*await service.role_view(code))

api/src/api/endpoints/admin/users.py

@@ -0,0 +1,66 @@
+"""Пользователи `/admin/users` (spec users-admin): список, создание, изменение/деактивация, замена ролей.
+
+Удаления нет (D-8): пользователь только деактивируется (`is_active=false`). Неизвестный id — 404.
+"""
+
+import uuid
+from typing import Annotated
+
+from fastapi import APIRouter, Query, status
+
+from src.api.dependencies.admin import UsersAdminServiceDep
+from src.schemas.admin import RolesSet, UserCreate, UserListQuery, UserOut, UserUpdate
+from src.schemas.common import Page, PageParams
+from src.schemas.directories import RegionalOfficeOut
+from src.services.users_admin import UserWithRoles
+
+router = APIRouter(prefix="/users")
+
+
+class UserListParams(UserListQuery, PageParams):
+    """Query списка пользователей: фильтры и пагинация одной моделью (FastAPI поддерживает одну Query-модель)."""
+
+
+def _out(view: UserWithRoles) -> UserOut:
+    user, office, roles = view
+    return UserOut(
+        id=user.id,
+        email=user.email,
+        full_name=user.full_name,
+        employee_no=user.employee_no,
+        regional_office=RegionalOfficeOut.model_validate(office),
+        business_segment=user.business_segment,
+        is_active=user.is_active,
+        roles=roles,
+        created_at=user.created_at,
+        updated_at=user.updated_at,
+    )
+
+
+@router.get("", response_model=Page[UserOut], summary="Список пользователей")
+async def list_users(service: UsersAdminServiceDep, query: Annotated[UserListParams, Query()]) -> Page[UserOut]:
+    """Пользователи с филиалом и ролями: поиск `q` по email/ФИО, фильтр `is_active`, сортировка по ФИО."""
+    page = PageParams(limit=query.limit, offset=query.offset)
+    rows, total = await service.list_users(query.q, query.is_active, page)
+    return Page[UserOut](items=[_out(row) for row in rows], total=total, limit=page.limit, offset=page.offset)
+
+
+@router.post("", response_model=UserOut, status_code=status.HTTP_201_CREATED, summary="Создать пользователя")
+async def create_user(data: UserCreate, service: UsersAdminServiceDep) -> UserOut:
+    """Email уникален без учёта регистра (409 `EMAIL_EXISTS`); роли — коды активных ролей (иначе 422 `UNKNOWN_ROLE`)."""
+    user = await service.create_user(data)
+    return _out(await service.user_view(user.id))
+
+
+@router.patch("/{user_id}", response_model=UserOut, summary="Изменить или деактивировать пользователя")
+async def update_user(user_id: uuid.UUID, data: UserUpdate, service: UsersAdminServiceDep) -> UserOut:
+    """Деактивация последнего активного обладателя `USERS_ADMIN` (и себя) — 409 `LAST_ADMIN`."""
+    user = await service.update_user(user_id, data)
+    return _out(await service.user_view(user.id))
+
+
+@router.put("/{user_id}/roles", response_model=UserOut, summary="Заменить роли пользователя")
+async def set_user_roles(user_id: uuid.UUID, data: RolesSet, service: UsersAdminServiceDep) -> UserOut:
+    """Набор ролей заменяется целиком; снятие последнего источника `USERS_ADMIN` — 409 `LAST_ADMIN`."""
+    await service.set_roles(user_id, data.roles)
+    return _out(await service.user_view(user_id))

api/src/api/router.py

@@ -1,12 +1,13 @@
 """Корневой роутер API `/api/v1` (R-6: эндпоинты — `src/api/endpoints/`).
 
-`build_api_router(settings)` собирает роутер под настройки: `/me` и `/directories/*` — всегда, dev-эмитент токенов —
-только при `AUTH_DEV_ISSUER and DEBUG`. Модульный `router` собран по настройкам процесса и подключается в `src/app.py`.
+`build_api_router(settings)` собирает роутер под настройки: `/me`, `/directories/*` и `/admin/*` — всегда,
+dev-эмитент токенов — только при `AUTH_DEV_ISSUER and DEBUG`. Модульный `router` собран по настройкам процесса и
+подключается в `src/app.py`.
 """
 
 from fastapi import APIRouter
 
-from src.api.endpoints import directories, me
+from src.api.endpoints import admin, directories, me
 from src.api.endpoints.dev_auth import build_dev_router
 from src.core.config import Settings, get_settings
 
@@ -18,6 +19,7 @@ def build_api_router(settings: Settings) -> APIRouter:
     api_router = APIRouter(prefix=API_PREFIX)
     api_router.include_router(me.router)
     api_router.include_router(directories.router)
+    api_router.include_router(admin.router)
     if settings.AUTH_DEV_ISSUER and settings.DEBUG:
         api_router.include_router(build_dev_router(settings))
     return api_router

api/src/repositories/users.py

@@ -1,17 +1,23 @@
 """Репозиторий пользователей: личность и эффективные права одним запросом (spec authn-authz, D-5, R-17).
 
 `effective_permissions(user) = ∪ permission.code по всем активным ролям пользователя` (CLAUDE.md: права — данные).
-Код роли в проверках не участвует: роли возвращаются только для показа (`GET /me`).
+Код роли в проверках не участвует: роли возвращаются только для показа (`GET /me`, список администратора).
+Запросы администрирования (spec users-admin): списки пользователей/ролей/прав, назначения, подсчёт обладателей права.
+Функции не делают `commit`: транзакцией запроса управляет `SessionDep`.
 """
 
+import uuid
+from collections.abc import Collection, Sequence
 from dataclasses import dataclass
+from typing import Any
 
-from sqlalchemy import and_, func, select
+from sqlalchemy import Select, and_, delete, func, or_, select
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import aliased
 
 from src.models.access import AppUser, Permission, Role, RolePermission, UserRole
 from src.models.directories import RegionalOffice
+from src.repositories.directories import LIKE_ESCAPE, escape_like
 
 # Достаточно двух строк, чтобы заметить неоднозначность; больше пользователей читать незачем
 _AMBIGUITY_PROBE_LIMIT = 2
@@ -71,3 +77,185 @@ async def load_principal_data(session: AsyncSession, email: str) -> UserWithAcce
         roles=tuple(sorted(roles)),
         permissions=permissions,
     )
+
+
+# --- Администрирование (spec users-admin, D-8) ----------------------------------------------------------------------
+
+
+def _filtered_users(stmt: Select[Any], q: str | None, is_active: bool | None) -> Select[Any]:
+    if q is not None and q.strip():
+        pattern = f"%{escape_like(q.strip())}%"
+        stmt = stmt.where(
+            or_(AppUser.email.ilike(pattern, escape=LIKE_ESCAPE), AppUser.full_name.ilike(pattern, escape=LIKE_ESCAPE))
+        )
+    if is_active is not None:
+        stmt = stmt.where(AppUser.is_active.is_(is_active))
+    return stmt
+
+
+async def list_users(
+    session: AsyncSession, *, q: str | None, is_active: bool | None, limit: int, offset: int
+) -> tuple[list[tuple[AppUser, RegionalOffice]], int]:
+    """Страница пользователей с филиалом (по ФИО, затем id) и общее число по фильтру."""
+    page_stmt = (
+        _filtered_users(select(AppUser, RegionalOffice), q, is_active)
+        .join(RegionalOffice, RegionalOffice.id == AppUser.regional_office_id)
+        .order_by(AppUser.full_name, AppUser.id)
+        .limit(limit)
+        .offset(offset)
+    )
+    total_stmt = _filtered_users(select(func.count(AppUser.id)), q, is_active)
+    rows = (await session.execute(page_stmt)).all()
+    total = (await session.execute(total_stmt)).scalar_one()
+    return [(user, office) for user, office in rows], total
+
+
+async def get_user(session: AsyncSession, user_id: uuid.UUID) -> AppUser | None:
+    """Пользователь по id."""
+    return await session.get(AppUser, user_id)
+
+
+async def email_taken(session: AsyncSession, normalized_email: str) -> bool:
+    """Есть ли пользователь с таким email без учёта регистра (`normalized_email` — уже `strip().lower()`)."""
+    stmt = select(AppUser.id).where(func.lower(AppUser.email) == normalized_email).limit(1)
+    return (await session.execute(stmt)).first() is not None
+
+
+async def role_codes_by_user(session: AsyncSession, user_ids: Sequence[uuid.UUID]) -> dict[uuid.UUID, list[str]]:
+    """Коды НАЗНАЧЕННЫХ ролей (включая неактивные роли) по пользователям, по алфавиту."""
+    if not user_ids:
+        return {}
+    stmt = (
+        select(UserRole.user_id, Role.code)
+        .join(Role, Role.id == UserRole.role_id)
+        .where(UserRole.user_id.in_(user_ids))
+        .order_by(Role.code)
+    )
+    result: dict[uuid.UUID, list[str]] = {user_id: [] for user_id in user_ids}
+    for user_id, code in (await session.execute(stmt)).all():
+        result[user_id].append(code)
+    return result
+
+
+async def user_roles(session: AsyncSession, user_id: uuid.UUID) -> list[tuple[UserRole, Role]]:
+    """Назначения ролей пользователя вместе с ролями."""
+    stmt = select(UserRole, Role).join(Role, Role.id == UserRole.role_id).where(UserRole.user_id == user_id)
+    return [(user_role, role) for user_role, role in (await session.execute(stmt)).all()]
+
+
+async def delete_user_roles(session: AsyncSession, user_id: uuid.UUID, role_ids: Collection[uuid.UUID]) -> None:
+    """Снимает с пользователя назначения указанных ролей."""
+    if role_ids:
+        await session.execute(delete(UserRole).where(UserRole.user_id == user_id, UserRole.role_id.in_(role_ids)))
+
+
+async def roles_by_codes(session: AsyncSession, codes: Collection[str]) -> list[Role]:
+    """Роли с указанными кодами (любого статуса)."""
+    if not codes:
+        return []
+    return list((await session.execute(select(Role).where(Role.code.in_(codes)))).scalars().all())
+
+
+async def get_role_by_code(session: AsyncSession, code: str) -> Role | None:
+    """Роль по коду."""
+    return (await session.execute(select(Role).where(Role.code == code))).scalar_one_or_none()
+
+
+async def list_roles_with_permissions(session: AsyncSession) -> list[tuple[Role, list[str]]]:
+    """Все роли (по коду) с кодами их прав (по алфавиту)."""
+    roles = list((await session.execute(select(Role).order_by(Role.code))).scalars().all())
+    stmt = (
+        select(RolePermission.role_id, Permission.code)
+        .join(Permission, Permission.id == RolePermission.permission_id)
+        .order_by(Permission.code)
+    )
+    codes: dict[uuid.UUID, list[str]] = {role.id: [] for role in roles}
+    for role_id, code in (await session.execute(stmt)).all():
+        codes.setdefault(role_id, []).append(code)
+    return [(role, codes[role.id]) for role in roles]
+
+
+async def role_permissions(session: AsyncSession, role_id: uuid.UUID) -> list[Permission]:
+    """Права роли (по коду)."""
+    stmt = (
+        select(Permission)
+        .join(RolePermission, RolePermission.permission_id == Permission.id)
+        .where(RolePermission.role_id == role_id)
+        .order_by(Permission.code)
+    )
+    return list((await session.execute(stmt)).scalars().all())
+
+
+async def delete_role_permissions(
+    session: AsyncSession, role_id: uuid.UUID, permission_ids: Collection[uuid.UUID]
+) -> None:
+    """Снимает с роли указанные права."""
+    if permission_ids:
+        await session.execute(
+            delete(RolePermission).where(
+                RolePermission.role_id == role_id, RolePermission.permission_id.in_(permission_ids)
+            )
+        )
+
+
+async def list_permissions(session: AsyncSession) -> list[Permission]:
+    """Справочник прав (по коду)."""
+    return list((await session.execute(select(Permission).order_by(Permission.code))).scalars().all())
+
+
+async def permissions_by_codes(session: AsyncSession, codes: Collection[str]) -> list[Permission]:
+    """Права с указанными кодами."""
+    if not codes:
+        return []
+    return list((await session.execute(select(Permission).where(Permission.code.in_(codes)))).scalars().all())
+
+
+async def lock_permission(session: AsyncSession, code: str) -> None:
+    """`SELECT … FOR NO KEY UPDATE` строки права: сериализует проверки «последнего обладателя права» между транзакциями.
+
+    Пока транзакция держит блокировку, конкурентная проверка того же права ждёт её завершения и затем считает уже
+    зафиксированное состояние (READ COMMITTED: каждый оператор видит данные, зафиксированные до его начала).
+    `NO KEY UPDATE` не мешает вставкам `role_permission` со ссылкой на это право (FK берёт `FOR KEY SHARE`).
+    """
+    await session.execute(select(Permission.id).where(Permission.code == code).with_for_update(key_share=True))
+
+
+async def roles_grant_permission(session: AsyncSession, role_ids: Collection[uuid.UUID], code: str) -> bool:
+    """Даёт ли хотя бы одна из АКТИВНЫХ ролей `role_ids` право `code`."""
+    if not role_ids:
+        return False
+    stmt = (
+        select(Role.id)
+        .join(RolePermission, RolePermission.role_id == Role.id)
+        .join(Permission, Permission.id == RolePermission.permission_id)
+        .where(Role.id.in_(role_ids), Role.is_active.is_(True), Permission.code == code)
+        .limit(1)
+    )
+    return (await session.execute(stmt)).first() is not None
+
+
+async def count_active_users_with_permission(
+    session: AsyncSession,
+    code: str,
+    *,
+    excluding_user: uuid.UUID | None = None,
+    excluding_role: uuid.UUID | None = None,
+) -> int:
+    """Число АКТИВНЫХ пользователей, которым АКТИВНЫЕ роли дают право `code` (как `effective_permissions`).
+
+    `excluding_user` — не считать этого пользователя; `excluding_role` — не считать право, полученное через эту роль
+    (пользователь учитывается, если право даёт другая его роль).
+    """
+    stmt = (
+        select(func.count(func.distinct(AppUser.id)))
+        .join(UserRole, UserRole.user_id == AppUser.id)
+        .join(Role, and_(Role.id == UserRole.role_id, Role.is_active.is_(True)))
+        .join(RolePermission, RolePermission.role_id == Role.id)
+        .join(Permission, Permission.id == RolePermission.permission_id)
+        .where(AppUser.is_active.is_(True), Permission.code == code)
+    )
+    if excluding_user is not None:
+        stmt = stmt.where(AppUser.id != excluding_user)
+    if excluding_role is not None:
+        stmt = stmt.where(Role.id != excluding_role)
+    return (await session.execute(stmt)).scalar_one()

api/src/schemas/admin.py

@@ -0,0 +1,150 @@
+"""Схемы API администрирования `/admin/*` (spec users-admin, design D-8): пользователи, роли, матрица прав.
+
+Поля — как в DBML v5.14 (`app_user`, `role`, `permission`); `external_id` через API не задаётся (появится при первом
+входе через РАИСА), поэтому тела запросов запрещают лишние поля.
+"""
+
+import datetime
+import uuid
+from typing import Self
+
+from pydantic import BaseModel, ConfigDict, EmailStr, Field, field_validator, model_validator
+
+from src.models.enums import BusinessSegment, PermissionAction, PermissionObject
+from src.schemas.directories import NAME_MAX_LENGTH, RegionalOfficeOut
+
+# Поля `app_user`, которые не могут быть NULL: явный `null` в PATCH — ошибка валидации, а не «очистить»
+_NOT_NULL_UPDATE_FIELDS = ("full_name", "regional_office_id", "is_active")
+
+
+def _strip_full_name(value: str) -> str:
+    stripped = value.strip()
+    if not stripped:
+        raise ValueError("ФИО не может быть пустым")
+    return stripped
+
+
+def _blank_to_none(value: str | None) -> str | None:
+    if value is None:
+        return None
+    return value.strip() or None
+
+
+class UserCreate(BaseModel):
+    """Создание пользователя: email уникален без учёта регистра (проверяет сервис), роли — коды активных ролей."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    email: EmailStr
+    full_name: str = Field(min_length=1, max_length=NAME_MAX_LENGTH)
+    regional_office_id: uuid.UUID
+    employee_no: str | None = None
+    business_segment: BusinessSegment | None = None
+    roles: list[str] = Field(default_factory=list)
+
+    @field_validator("full_name")
+    @classmethod
+    def _full_name(cls, value: str) -> str:
+        return _strip_full_name(value)
+
+    @field_validator("employee_no")
+    @classmethod
+    def _employee_no(cls, value: str | None) -> str | None:
+        return _blank_to_none(value)
+
+
+class UserUpdate(BaseModel):
+    """Изменение пользователя; «поле передано» — по `model_fields_set`.
+
+    `employee_no` и `business_segment` можно очистить (`null`); `full_name`, `regional_office_id`, `is_active` — нет.
+    """
+
+    model_config = ConfigDict(extra="forbid")
+
+    full_name: str | None = Field(default=None, min_length=1, max_length=NAME_MAX_LENGTH)
+    employee_no: str | None = None
+    regional_office_id: uuid.UUID | None = None
+    business_segment: BusinessSegment | None = None
+    is_active: bool | None = None
+
+    @field_validator("full_name")
+    @classmethod
+    def _full_name(cls, value: str | None) -> str | None:
+        return None if value is None else _strip_full_name(value)
+
+    @field_validator("employee_no")
+    @classmethod
+    def _employee_no(cls, value: str | None) -> str | None:
+        return _blank_to_none(value)
+
+    @model_validator(mode="after")
+    def _no_null_for_required(self) -> Self:
+        nulls = [
+            name for name in _NOT_NULL_UPDATE_FIELDS if name in self.model_fields_set and getattr(self, name) is None
+        ]
+        if nulls:
+            raise ValueError(f"поля не могут быть null: {', '.join(nulls)}")
+        return self
+
+
+class UserOut(BaseModel):
+    """Пользователь для администратора: филиал целиком и коды назначенных ролей (по алфавиту)."""
+
+    model_config = ConfigDict(use_enum_values=True)
+
+    id: uuid.UUID
+    email: str | None
+    full_name: str
+    employee_no: str | None
+    regional_office: RegionalOfficeOut
+    business_segment: BusinessSegment | None
+    is_active: bool
+    roles: list[str]
+    created_at: datetime.datetime | None
+    updated_at: datetime.datetime | None
+
+
+class UserListQuery(BaseModel):
+    """Фильтры списка пользователей: `q` — подстрока email или ФИО без учёта регистра, `is_active`."""
+
+    q: str | None = None
+    is_active: bool | None = None
+
+
+class RolesSet(BaseModel):
+    """Новый набор ролей пользователя (коды); заменяет прежний целиком."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    roles: list[str]
+
+
+class RoleOut(BaseModel):
+    """Роль с кодами её прав (по алфавиту)."""
+
+    code: str
+    name: str
+    description: str | None
+    is_system: bool
+    is_active: bool
+    permissions: list[str]
+
+
+class PermissionOut(BaseModel):
+    """Право из справочника (только чтение: новые коды не создаются)."""
+
+    model_config = ConfigDict(from_attributes=True, use_enum_values=True)
+
+    code: str
+    object_type: PermissionObject
+    action: PermissionAction
+    name: str
+    description: str | None
+
+
+class PermissionsSet(BaseModel):
+    """Новый набор прав роли (коды); заменяет прежний целиком."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    permissions: list[str]

api/src/services/users_admin.py

@@ -0,0 +1,302 @@
+"""Сервис администрирования (spec users-admin, audit-trail; design D-8; `05-domain-rules.md` §10).
+
+Пользователи (создание, изменение, деактивация — без удаления), назначение ролей, матрица «роль × право». Каждая
+мутация пишет `audit_event` в транзакции запроса (`USER_CHANGE` / `USER_ROLE_CHANGE` — раздел `USERS`,
+`ROLE_PERMISSION_CHANGE` — `ACCESS`); повтор того же состояния ничего не пишет.
+
+Защита `LAST_ADMIN`: после операции у активных пользователей должно остаться право `USERS_ADMIN`, полученное через
+активную роль (как в `effective_permissions`). Проверяется на деактивации пользователя, замене его ролей и снятии
+права с роли. Код роли не читается — только право.
+
+Гонки: мутируемая строка (`app_user` / `role`) блокируется `FOR NO KEY UPDATE`, затем перед подсчётом — строка права
+`USERS_ADMIN` (`repo.lock_permission`): проверки «последнего администратора» выполняются по очереди, вторая видит
+зафиксированный результат первой. Порядок блокировок всегда «объект → право». Именно `NO KEY UPDATE`, а не
+`FOR UPDATE`: вставки `audit_event.user_id`, `user_role.role_id`, `role_permission.granted_by` берут на родительские
+строки `FOR KEY SHARE` (проверка FK), который `FOR UPDATE` блокирует — два администратора, правящие друг друга, ловили
+бы взаимную блокировку (deadlock) на вставке аудита.
+"""
+
+import uuid
+from collections.abc import Iterable
+from typing import Any
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.api.dependencies.audit import RequestAudit
+from src.common.clock import utc_now
+from src.common.exceptions import ConflictError, DomainValidationError, NotFoundError
+from src.models.access import AppUser, Permission, Role, RolePermission, UserRole
+from src.models.directories import RegionalOffice
+from src.models.enums import AuditAction, AuditSection
+from src.repositories import directories as directories_repo
+from src.repositories import users as repo
+from src.schemas.admin import UserCreate, UserUpdate
+from src.schemas.common import PageParams
+
+USERS_ADMIN = "USERS_ADMIN"  # право, которое не должно пропасть у всех активных пользователей (D-8)
+USER_OBJECT = "app_user"
+ROLE_OBJECT = "role"
+USER_NOT_FOUND = "Пользователь не найден"
+ROLE_NOT_FOUND = "Роль не найдена"
+LAST_ADMIN_MESSAGE = "Нельзя оставить систему без активного пользователя с правом USERS_ADMIN"
+UPDATE_FIELDS = ("full_name", "employee_no", "regional_office_id", "business_segment", "is_active")
+
+UserWithRoles = tuple[AppUser, RegionalOffice, list[str]]
+
+
+def _unique(codes: Iterable[str]) -> list[str]:
+    """Коды без дубликатов в порядке первого появления."""
+    return list(dict.fromkeys(codes))
+
+
+class UsersAdminService:
+    """Операции администратора; права (`USERS_ADMIN`) проверяет маршрут, сервис — инварианты и аудит."""
+
+    def __init__(self, session: AsyncSession, audit: RequestAudit, actor_id: uuid.UUID) -> None:
+        self._session = session
+        self._audit = audit
+        self._actor_id = actor_id
+
+    # --- пользователи ---------------------------------------------------------------------------------------------
+
+    async def list_users(
+        self, q: str | None, is_active: bool | None, page: PageParams
+    ) -> tuple[list[UserWithRoles], int]:
+        """Страница пользователей с филиалом и кодами назначенных ролей; поиск `q` по email/ФИО."""
+        rows, total = await repo.list_users(
+            self._session, q=q, is_active=is_active, limit=page.limit, offset=page.offset
+        )
+        roles = await repo.role_codes_by_user(self._session, [user.id for user, _ in rows])
+        return [(user, office, roles[user.id]) for user, office in rows], total
+
+    async def user_view(self, user_id: uuid.UUID) -> UserWithRoles:
+        """Пользователь с филиалом и ролями (для ответа); нет — 404."""
+        user = await repo.get_user(self._session, user_id)
+        if user is None:
+            raise NotFoundError(USER_NOT_FOUND)
+        office = await self._session.get(RegionalOffice, user.regional_office_id)
+        if office is None:  # FK NOT NULL: недостижимо при целостной БД
+            raise NotFoundError("Филиал пользователя не найден")
+        roles = await repo.role_codes_by_user(self._session, [user.id])
+        return user, office, roles[user.id]
+
+    async def create_user(self, data: UserCreate) -> AppUser:
+        """Новый пользователь с ролями; email уникален без учёта регистра (409 `EMAIL_EXISTS`), хранится как введён."""
+        email = str(data.email).strip()
+        if await repo.email_taken(self._session, email.lower()):
+            raise ConflictError("EMAIL_EXISTS", "Пользователь с таким email уже существует")
+        await self._check_office(data.regional_office_id)
+        codes = _unique(data.roles)
+        roles = await self._active_roles(codes)
+        now = utc_now()
+        user = AppUser(
+            email=email,
+            full_name=data.full_name,
+            employee_no=data.employee_no,
+            regional_office_id=data.regional_office_id,
+            business_segment=data.business_segment,
+            is_active=True,
+            created_at=now,
+            updated_at=now,
+        )
+        self._session.add(user)
+        await self._session.flush()
+        self._session.add_all(UserRole(user_id=user.id, role_id=role.id) for role in roles)
+        await self._session.flush()
+        await self._audit.record(
+            action=AuditAction.USER_CHANGE,
+            section=AuditSection.USERS,
+            object_type=USER_OBJECT,
+            object_id=user.id,
+            after_value={
+                "email": user.email,
+                "full_name": user.full_name,
+                "employee_no": user.employee_no,
+                "regional_office_id": user.regional_office_id,
+                "business_segment": user.business_segment,
+                "roles": codes,
+            },
+        )
+        return user
+
+    async def update_user(self, user_id: uuid.UUID, data: UserUpdate) -> AppUser:
+        """Изменение полей и деактивация; без фактических изменений ничего не пишется.
+
+        Деактивация последнего активного обладателя `USERS_ADMIN` (в том числе самого себя) — 409 `LAST_ADMIN`.
+        """
+        user = await self._lock_user(user_id)
+        changes: dict[str, Any] = {}
+        for field in UPDATE_FIELDS:
+            if field in data.model_fields_set and getattr(data, field) != getattr(user, field):
+                changes[field] = getattr(data, field)
+        if not changes:
+            return user
+        if "regional_office_id" in changes:
+            await self._check_office(changes["regional_office_id"])
+        if changes.get("is_active") is False:
+            await self._ensure_admin_remains(excluding_user=user.id)
+        before = {field: getattr(user, field) for field in changes}
+        for field, value in changes.items():
+            setattr(user, field, value)
+        user.updated_at = utc_now()
+        await self._session.flush()
+        await self._audit.record(
+            action=AuditAction.USER_CHANGE,
+            section=AuditSection.USERS,
+            object_type=USER_OBJECT,
+            object_id=user.id,
+            before_value=before,
+            after_value=changes,
+            changed_field=next(iter(changes)) if len(changes) == 1 else None,
+        )
+        return user
+
+    async def set_roles(self, user_id: uuid.UUID, codes: list[str]) -> list[str]:
+        """Заменяет набор ролей пользователя; возвращает новые коды по алфавиту.
+
+        Неизвестный или неактивный код — 422 `UNKNOWN_ROLE`; если после замены ни у одного активного пользователя не
+        останется `USERS_ADMIN` — 409 `LAST_ADMIN`. Аудит `USER_ROLE_CHANGE`: before — прежние коды, after — новые
+        в порядке запроса.
+        """
+        user = await self._lock_user(user_id)
+        new_codes = _unique(codes)
+        new_roles = await self._active_roles(new_codes)
+        current = await repo.user_roles(self._session, user.id)
+        old_ids = {role.id for _, role in current}
+        new_ids = {role.id for role in new_roles}
+        if old_ids == new_ids:
+            return sorted(new_codes)
+        removed = old_ids - new_ids
+        if removed:
+            await self._ensure_admin_remains(excluding_user=user.id, kept_roles=new_ids if user.is_active else None)
+        await repo.delete_user_roles(self._session, user.id, removed)
+        self._session.add_all(UserRole(user_id=user.id, role_id=role_id) for role_id in new_ids - old_ids)
+        await self._session.flush()
+        await self._audit.record(
+            action=AuditAction.USER_ROLE_CHANGE,
+            section=AuditSection.USERS,
+            object_type=USER_OBJECT,
+            object_id=user.id,
+            before_value={"roles": sorted(role.code for _, role in current)},
+            after_value={"roles": new_codes},
+            changed_field="roles",
+        )
+        return sorted(new_codes)
+
+    # --- роли и права ---------------------------------------------------------------------------------------------
+
+    async def list_roles(self) -> list[tuple[Role, list[str]]]:
+        """Все роли (по коду) с кодами прав."""
+        return await repo.list_roles_with_permissions(self._session)
+
+    async def list_permissions(self) -> list[Permission]:
+        """Справочник прав (по коду)."""
+        return await repo.list_permissions(self._session)
+
+    async def role_view(self, code: str) -> tuple[Role, list[str]]:
+        """Роль с кодами прав; нет — 404."""
+        role = await repo.get_role_by_code(self._session, code)
+        if role is None:
+            raise NotFoundError(ROLE_NOT_FOUND)
+        return role, [permission.code for permission in await repo.role_permissions(self._session, role.id)]
+
+    async def set_role_permissions(self, code: str, permission_codes: list[str]) -> list[str]:
+        """Заменяет набор прав роли; возвращает новые коды по алфавиту.
+
+        Неизвестная роль — 404; неизвестный код права — 422 `UNKNOWN_PERMISSION` (новые права не создаются); снятие
+        `USERS_ADMIN`, после которого его не останется ни у одного активного пользователя, — 409 `LAST_ADMIN`.
+        """
+        stmt = (
+            select(Role)
+            .where(Role.code == code)
+            .with_for_update(key_share=True)
+            .execution_options(populate_existing=True)
+        )
+        role = (await self._session.execute(stmt)).scalar_one_or_none()
+        if role is None:
+            raise NotFoundError(ROLE_NOT_FOUND)
+        new_codes = _unique(permission_codes)
+        found = {
+            permission.code: permission for permission in await repo.permissions_by_codes(self._session, new_codes)
+        }
+        unknown = [item for item in new_codes if item not in found]
+        if unknown:
+            raise DomainValidationError(
+                "UNKNOWN_PERMISSION", f"Неизвестные коды прав: {', '.join(unknown)}", permissions=unknown
+            )
+        current = {permission.code: permission for permission in await repo.role_permissions(self._session, role.id)}
+        if set(current) == set(found):
+            return sorted(found)
+        removed = set(current) - set(found)
+        if USERS_ADMIN in removed:
+            await self._ensure_admin_remains(excluding_role=role.id)
+        await repo.delete_role_permissions(self._session, role.id, {current[item].id for item in removed})
+        now = utc_now()
+        self._session.add_all(
+            RolePermission(role_id=role.id, permission_id=found[item].id, granted_by=self._actor_id, granted_at=now)
+            for item in set(found) - set(current)
+        )
+        await self._session.flush()
+        await self._audit.record(
+            action=AuditAction.ROLE_PERMISSION_CHANGE,
+            section=AuditSection.ACCESS,
+            object_type=ROLE_OBJECT,
+            object_id=role.id,
+            before_value={"permissions": sorted(current)},
+            after_value={"permissions": sorted(found)},
+            changed_field="permissions",
+        )
+        return sorted(found)
+
+    # --- проверки -------------------------------------------------------------------------------------------------
+
+    async def _lock_user(self, user_id: uuid.UUID) -> AppUser:
+        """Пользователь под `FOR NO KEY UPDATE` (конкурентные правки одного пользователя идут по очереди); нет — 404."""
+        stmt = (
+            select(AppUser)
+            .where(AppUser.id == user_id)
+            .with_for_update(key_share=True)
+            .execution_options(populate_existing=True)
+        )
+        user = (await self._session.execute(stmt)).scalar_one_or_none()
+        if user is None:
+            raise NotFoundError(USER_NOT_FOUND)
+        return user
+
+    async def _active_roles(self, codes: list[str]) -> list[Role]:
+        """Активные роли по кодам; неизвестные и неактивные — 422 `UNKNOWN_ROLE` со списком кодов."""
+        roles = {role.code: role for role in await repo.roles_by_codes(self._session, codes) if role.is_active}
+        unknown = [code for code in codes if code not in roles]
+        if unknown:
+            raise DomainValidationError(
+                "UNKNOWN_ROLE", f"Неизвестные или неактивные роли: {', '.join(unknown)}", roles=unknown
+            )
+        return [roles[code] for code in codes]
+
+    async def _check_office(self, office_id: uuid.UUID) -> None:
+        """Филиал из тела запроса должен существовать — 422 `OFFICE_NOT_FOUND` (как у ГК/ЮЛ, R-8)."""
+        if await directories_repo.get_regional_office(self._session, office_id) is None:
+            raise DomainValidationError("OFFICE_NOT_FOUND", "Филиал не найден")
+
+    async def _ensure_admin_remains(
+        self,
+        *,
+        excluding_user: uuid.UUID | None = None,
+        excluding_role: uuid.UUID | None = None,
+        kept_roles: set[uuid.UUID] | None = None,
+    ) -> None:
+        """409 `LAST_ADMIN`, если после операции ни у одного активного пользователя не останется `USERS_ADMIN`.
+
+        Сначала блокируется строка права (см. докстринг модуля), затем — уже по зафиксированному состоянию — считаются
+        ДРУГИЕ обладатели права: без `excluding_user` или без права через `excluding_role`. `kept_roles` — роли, которые
+        останутся у активного `excluding_user`: если одна из них даёт `USERS_ADMIN`, право у него сохраняется.
+        """
+        await repo.lock_permission(self._session, USERS_ADMIN)
+        if kept_roles is not None and await repo.roles_grant_permission(self._session, kept_roles, USERS_ADMIN):
+            return
+        others = await repo.count_active_users_with_permission(
+            self._session, USERS_ADMIN, excluding_user=excluding_user, excluding_role=excluding_role
+        )
+        if others == 0:
+            raise ConflictError("LAST_ADMIN", LAST_ADMIN_MESSAGE)

api/tests/test_api_admin.py

@@ -0,0 +1,575 @@
+"""API администрирования `/admin/*` (spec users-admin, audit-trail; design D-8; Review Focus 5 фазы 2).
+
+Пользователи (создание, изменение, деактивация), назначение ролей, матрица «роль × право», аудит и защита `LAST_ADMIN`:
+последнее право `USERS_ADMIN` у активных пользователей нельзя снять ни деактивацией, ни заменой ролей, ни снятием
+права с роли. Новые пользователи — с доменом `example.com`: `EmailStr` не принимает зарезервированный `.local`
+dev-пользователей.
+"""
+
+import asyncio
+import uuid
+from collections.abc import Sequence
+from typing import Any
+
+import httpx
+import pytest
+from sqlalchemy import func, select
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from src.api.dependencies.audit import RequestAudit
+from src.common.exceptions import DomainError
+from src.models.access import AppUser, Permission, Role, RolePermission, UserRole
+from src.models.audit_ex import AuditEvent
+from src.models.directories import RegionalOffice
+from src.models.enums import AuditAction, AuditSection
+from src.repositories import users as users_repo
+from src.services.audit import AuditService
+from src.services.users_admin import UsersAdminService
+from tests.conftest import AuthClientFactory
+from tests.factories import make_user
+
+HTTP_OK = 200
+HTTP_CREATED = 201
+HTTP_FORBIDDEN = 403
+HTTP_NOT_FOUND = 404
+HTTP_CONFLICT = 409
+HTTP_UNPROCESSABLE = 422
+
+ADMIN = "admin@armka.local"
+USERS = "/admin/users"
+ROSTOV = "Ростовский региональный филиал"
+HEAD_OFFICE = "Головной офис"
+PERMISSIONS_TOTAL = 16  # seed: 16 кодов прав (§10)
+EXPERT_PERMISSIONS = ["CONSOLIDATION_VIEW", "DIRECTORY_VIEW", "DOCUMENT_VIEW", "EX_VIEW", "KFO_VIEW"]
+
+
+async def _office(session: AsyncSession, name: str) -> RegionalOffice:
+    return (await session.execute(select(RegionalOffice).where(RegionalOffice.name == name))).scalar_one()
+
+
+async def _user_id(session: AsyncSession, email: str) -> uuid.UUID:
+    return (await session.execute(select(AppUser.id).where(AppUser.email == email))).scalar_one()
+
+
+async def _role_codes(session: AsyncSession, user_id: uuid.UUID) -> list[str]:
+    stmt = select(Role.code).join(UserRole, UserRole.role_id == Role.id).where(UserRole.user_id == user_id)
+    return sorted((await session.execute(stmt)).scalars().all())
+
+
+async def _events(session: AsyncSession, action: AuditAction) -> Sequence[AuditEvent]:
+    stmt = select(AuditEvent).where(AuditEvent.action == action).order_by(AuditEvent.occurred_at, AuditEvent.id)
+    return (await session.execute(stmt)).scalars().all()
+
+
+async def _create(client: httpx.AsyncClient, **body: Any) -> dict[str, Any]:
+    response = await client.post(USERS, json=body)
+    assert response.status_code == HTTP_CREATED, response.text
+    created: dict[str, Any] = response.json()
+    return created
+
+
+def _detail_code(response: httpx.Response) -> str:
+    code: str = response.json()["detail"]["code"]
+    return code
+
+
+# --- Пользователи ------------------------------------------------------------------------------------------------
+
+
+@pytest.mark.integration
+async def test_create_user_with_role_gives_role_permissions_in_me(
+    auth_client: AuthClientFactory, seeded: AsyncSession
+) -> None:
+    admin = await auth_client(ADMIN)
+    rostov = await _office(seeded, ROSTOV)
+    created = await _create(
+        admin,
+        email="  Ivan.Petrov@Example.com ",
+        full_name="Иван Петров",
+        regional_office_id=str(rostov.id),
+        employee_no="E-1",
+        business_segment="DKB",
+        roles=["KA_RF"],
+    )
+    assert created["email"] == "Ivan.Petrov@example.com"  # регистр локальной части сохранён (EmailStr: домен строчными)
+    assert created["roles"] == ["KA_RF"]
+    assert created["is_active"] is True
+    assert created["regional_office"] == {
+        "id": str(rostov.id),
+        "name": ROSTOV,
+        "is_head_office": False,
+        "is_active": True,
+    }
+    assert set(created) == {
+        "id",
+        "email",
+        "full_name",
+        "employee_no",
+        "regional_office",
+        "business_segment",
+        "is_active",
+        "roles",
+        "created_at",
+        "updated_at",
+    }
+    user_id = uuid.UUID(created["id"])
+    assert await _role_codes(seeded, user_id) == ["KA_RF"]
+    user = await seeded.get(AppUser, user_id)
+    assert user is not None and user.external_id is None and user.created_at is not None
+
+    events = await _events(seeded, AuditAction.USER_CHANGE)
+    assert len(events) == 1
+    event = events[0]
+    assert (event.section, event.object_type, event.object_id) == (AuditSection.USERS, "app_user", user_id)
+    assert event.user_id == await _user_id(seeded, ADMIN)
+    assert event.after_value == {
+        "email": "Ivan.Petrov@example.com",
+        "full_name": "Иван Петров",
+        "employee_no": "E-1",
+        "regional_office_id": str(rostov.id),
+        "business_segment": "DKB",
+        "roles": ["KA_RF"],
+    }
+
+    reference = (await (await auth_client("ka_rf@armka.local")).get("/me")).json()
+    me = await (await auth_client("ivan.petrov@example.com")).get("/me")
+    assert me.status_code == HTTP_OK, me.text
+    assert me.json()["roles"] == ["KA_RF"]
+    assert me.json()["permissions"] == reference["permissions"]
+
+
+@pytest.mark.integration
+async def test_create_user_email_in_other_case_conflicts(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
+    admin = await auth_client(ADMIN)
+    office_id = str((await _office(seeded, HEAD_OFFICE)).id)
+    await _create(admin, email="anna@example.com", full_name="Анна", regional_office_id=office_id)
+    response = await admin.post(
+        USERS, json={"email": "ANNA@Example.COM", "full_name": "Анна 2", "regional_office_id": office_id}
+    )
+    assert response.status_code == HTTP_CONFLICT
+    assert _detail_code(response) == "EMAIL_EXISTS"
+    total = (await seeded.execute(select(func.count()).where(func.lower(AppUser.email) == "anna@example.com"))).scalar()
+    assert total == 1
+
+
+@pytest.mark.integration
+async def test_create_user_validation(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
+    admin = await auth_client(ADMIN)
+    office_id = str((await _office(seeded, HEAD_OFFICE)).id)
+    base = {"email": "v@example.com", "full_name": "В", "regional_office_id": office_id}
+
+    unknown_role = await admin.post(USERS, json={**base, "roles": ["KA_RF", "NO_SUCH_ROLE"]})
+    assert unknown_role.status_code == HTTP_UNPROCESSABLE
+    assert unknown_role.json()["detail"]["code"] == "UNKNOWN_ROLE"
+    assert unknown_role.json()["detail"]["roles"] == ["NO_SUCH_ROLE"]
+
+    unknown_office = await admin.post(USERS, json={**base, "regional_office_id": str(uuid.uuid4())})
+    assert unknown_office.status_code == HTTP_UNPROCESSABLE
+    assert _detail_code(unknown_office) == "OFFICE_NOT_FOUND"
+
+    for body in (
+        {**base, "external_id": "raisa-1"},  # external_id через API не задаётся
+        {**base, "email": "not-an-email"},
+        {**base, "full_name": "   "},
+    ):
+        response = await admin.post(USERS, json=body)
+        assert response.status_code == HTTP_UNPROCESSABLE, body
+    assert (await seeded.execute(select(AppUser).where(AppUser.email == "v@example.com"))).first() is None
+    assert await _events(seeded, AuditAction.USER_CHANGE) == []
+
+
+@pytest.mark.integration
+async def test_list_users_search_filter_and_page(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
+    office = await _office(seeded, ROSTOV)
+    await make_user(seeded, email="Zoya_Search@example.com", office=office, roles=["KA_RF", "EXPERT"], full_name="Зоя")
+    await make_user(seeded, email="off@example.com", office=office, full_name="Зоя Отключённая", is_active=False)
+    admin = await auth_client(ADMIN)
+
+    everyone = (await admin.get(USERS)).json()
+    assert set(everyone) == {"items", "total", "limit", "offset"}
+    assert everyone["total"] == len(everyone["items"]) >= 9  # noqa: PLR2004 - 7 dev-пользователей + 2 новых
+
+    by_email = (await admin.get(USERS, params={"q": "zoya_s"})).json()  # `_` ищется буквально, без регистра
+    assert [item["email"] for item in by_email["items"]] == ["Zoya_Search@example.com"]
+    assert by_email["items"][0]["roles"] == ["EXPERT", "KA_RF"]
+    assert by_email["items"][0]["regional_office"]["name"] == ROSTOV
+
+    by_name = (await admin.get(USERS, params={"q": "зоя"})).json()
+    assert by_name["total"] == 2  # noqa: PLR2004
+    inactive = (await admin.get(USERS, params={"q": "зоя", "is_active": "false"})).json()
+    assert [item["email"] for item in inactive["items"]] == ["off@example.com"]
+    assert inactive["items"][0]["roles"] == []
+
+    page = (await admin.get(USERS, params={"limit": 2, "offset": 1})).json()
+    assert (len(page["items"]), page["limit"], page["offset"], page["total"]) == (2, 2, 1, everyone["total"])
+    assert page["items"] == everyone["items"][1:3]
+
+
+@pytest.mark.integration
+async def test_patch_user_fields_audit_and_noop(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
+    rostov = await _office(seeded, ROSTOV)
+    head = await _office(seeded, HEAD_OFFICE)
+    user = await make_user(seeded, email="p@example.com", office=rostov, roles=["KA_RF"], full_name="Пётр")
+    admin = await auth_client(ADMIN)
+
+    response = await admin.patch(
+        f"{USERS}/{user.id}", json={"full_name": "Пётр Петров", "regional_office_id": str(head.id)}
+    )
+    assert response.status_code == HTTP_OK, response.text
+    body = response.json()
+    assert body["full_name"] == "Пётр Петров"
+    assert body["regional_office"]["id"] == str(head.id)
+    assert body["roles"] == ["KA_RF"]
+    events = await _events(seeded, AuditAction.USER_CHANGE)
+    assert len(events) == 1
+    assert (events[0].section, events[0].object_type, events[0].object_id) == (AuditSection.USERS, "app_user", user.id)
+    assert events[0].before_value == {"full_name": "Пётр", "regional_office_id": str(rostov.id)}
+    assert events[0].after_value == {"full_name": "Пётр Петров", "regional_office_id": str(head.id)}
+    assert events[0].changed_field is None
+
+    single = await admin.patch(f"{USERS}/{user.id}", json={"employee_no": "E-7"})
+    assert single.status_code == HTTP_OK
+    last = (await _events(seeded, AuditAction.USER_CHANGE))[-1]
+    assert (last.changed_field, last.before_value, last.after_value) == (
+        "employee_no",
+        {"employee_no": None},
+        {"employee_no": "E-7"},
+    )
+
+    noop = await admin.patch(f"{USERS}/{user.id}", json={"full_name": "Пётр Петров", "employee_no": "E-7"})
+    assert noop.status_code == HTTP_OK
+    assert len(await _events(seeded, AuditAction.USER_CHANGE)) == 2  # noqa: PLR2004 - no-op не пишет аудит
+
+    for bad in ({"email": "x@example.com"}, {"full_name": None}, {"is_active": None}, {"regional_office_id": None}):
+        assert (await admin.patch(f"{USERS}/{user.id}", json=bad)).status_code == HTTP_UNPROCESSABLE, bad
+    office_missing = await admin.patch(f"{USERS}/{user.id}", json={"regional_office_id": str(uuid.uuid4())})
+    assert office_missing.status_code == HTTP_UNPROCESSABLE
+    assert _detail_code(office_missing) == "OFFICE_NOT_FOUND"
+    assert (await admin.patch(f"{USERS}/{uuid.uuid4()}", json={"full_name": "Х"})).status_code == HTTP_NOT_FOUND
+
+
+@pytest.mark.integration
+async def test_deactivated_user_loses_access(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
+    user_id = await _user_id(seeded, "expert@armka.local")
+    expert = await auth_client("expert@armka.local")
+    admin = await auth_client(ADMIN)
+    response = await admin.patch(f"{USERS}/{user_id}", json={"is_active": False})
+    assert response.status_code == HTTP_OK
+    assert response.json()["is_active"] is False
+    assert (await expert.get("/me")).status_code == HTTP_FORBIDDEN
+    event = (await _events(seeded, AuditAction.USER_CHANGE))[-1]
+    assert (event.changed_field, event.before_value, event.after_value) == (
+        "is_active",
+        {"is_active": True},
+        {"is_active": False},
+    )
+
+
+# --- Назначение ролей --------------------------------------------------------------------------------------------
+
+
+@pytest.mark.integration
+async def test_put_roles_replaces_set_with_audit(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
+    user = await make_user(seeded, email="r@example.com", office=await _office(seeded, ROSTOV), roles=["KA_RF"])
+    admin = await auth_client(ADMIN)
+
+    response = await admin.put(f"{USERS}/{user.id}/roles", json={"roles": ["KA_GO", "EXPERT", "KA_GO"]})
+    assert response.status_code == HTTP_OK, response.text
+    assert response.json()["roles"] == ["EXPERT", "KA_GO"]
+    assert await _role_codes(seeded, user.id) == ["EXPERT", "KA_GO"]
+    rows = (await seeded.execute(select(func.count()).where(UserRole.user_id == user.id))).scalar()
+    assert rows == 2  # noqa: PLR2004
+
+    events = await _events(seeded, AuditAction.USER_ROLE_CHANGE)
+    assert len(events) == 1
+    event = events[0]
+    assert (event.section, event.object_type, event.object_id) == (AuditSection.USERS, "app_user", user.id)
+    assert event.before_value == {"roles": ["KA_RF"]}
+    assert event.after_value == {"roles": ["KA_GO", "EXPERT"]}
+
+    same = await admin.put(f"{USERS}/{user.id}/roles", json={"roles": ["EXPERT", "KA_GO"]})
+    assert same.status_code == HTTP_OK
+    assert len(await _events(seeded, AuditAction.USER_ROLE_CHANGE)) == 1  # тот же набор — без аудита
+
+    cleared = await admin.put(f"{USERS}/{user.id}/roles", json={"roles": []})
+    assert cleared.status_code == HTTP_OK
+    assert await _role_codes(seeded, user.id) == []
+
+
+@pytest.mark.integration
+async def test_put_roles_rejects_unknown_and_inactive_role(
+    auth_client: AuthClientFactory, seeded: AsyncSession
+) -> None:
+    user = await make_user(seeded, email="u@example.com", office=await _office(seeded, ROSTOV), roles=["KA_RF"])
+    seeded.add(Role(code="RETIRED", name="Упразднённая", is_system=False, is_active=False))
+    await seeded.flush()
+    admin = await auth_client(ADMIN)
+
+    for codes, bad in ((["KA_GO", "NOPE"], ["NOPE"]), (["RETIRED"], ["RETIRED"])):
+        response = await admin.put(f"{USERS}/{user.id}/roles", json={"roles": codes})
+        assert response.status_code == HTTP_UNPROCESSABLE
+        assert response.json()["detail"]["code"] == "UNKNOWN_ROLE"
+        assert response.json()["detail"]["roles"] == bad
+    assert await _role_codes(seeded, user.id) == ["KA_RF"]
+    missing = await admin.put(f"{USERS}/{uuid.uuid4()}/roles", json={"roles": ["KA_GO"]})
+    assert missing.status_code == HTTP_NOT_FOUND
+
+
+# --- LAST_ADMIN --------------------------------------------------------------------------------------------------
+
+
+@pytest.mark.integration
+async def test_last_admin_cannot_drop_own_admin_role(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
+    admin_id = await _user_id(seeded, ADMIN)
+    admin = await auth_client(ADMIN)
+    response = await admin.put(f"{USERS}/{admin_id}/roles", json={"roles": ["EXPERT"]})
+    assert response.status_code == HTTP_CONFLICT
+    assert _detail_code(response) == "LAST_ADMIN"
+    assert await _role_codes(seeded, admin_id) == ["ADMIN"]
+    assert await _events(seeded, AuditAction.USER_ROLE_CHANGE) == []
+
+
+@pytest.mark.integration
+async def test_inactive_user_or_inactive_role_is_not_an_admin_source(
+    auth_client: AuthClientFactory, seeded: AsyncSession
+) -> None:
+    head = await _office(seeded, HEAD_OFFICE)
+    await make_user(seeded, email="sleeping.admin@example.com", office=head, roles=["ADMIN"], is_active=False)
+    retired = Role(code="RETIRED_ADMIN", name="Бывшая админская", is_system=False, is_active=False)
+    seeded.add(retired)
+    await seeded.flush()
+    users_admin = (await seeded.execute(select(Permission.id).where(Permission.code == "USERS_ADMIN"))).scalar_one()
+    seeded.add(RolePermission(role_id=retired.id, permission_id=users_admin))
+    await make_user(seeded, email="retired.admin@example.com", office=head, roles=["RETIRED_ADMIN"])
+
+    admin_id = await _user_id(seeded, ADMIN)
+    admin = await auth_client(ADMIN)
+    response = await admin.put(f"{USERS}/{admin_id}/roles", json={"roles": ["EXPERT"]})
+    assert response.status_code == HTTP_CONFLICT
+    assert _detail_code(response) == "LAST_ADMIN"
+
+
+@pytest.mark.integration
+async def test_second_admin_allows_dropping_own_admin_role(
+    auth_client: AuthClientFactory, seeded: AsyncSession
+) -> None:
+    await make_user(
+        seeded, email="second.admin@example.com", office=await _office(seeded, HEAD_OFFICE), roles=["ADMIN"]
+    )
+    admin_id = await _user_id(seeded, ADMIN)
+    admin = await auth_client(ADMIN)
+    response = await admin.put(f"{USERS}/{admin_id}/roles", json={"roles": ["EXPERT"]})
+    assert response.status_code == HTTP_OK, response.text
+    assert await _role_codes(seeded, admin_id) == ["EXPERT"]
+    assert (await admin.get("/admin/roles")).status_code == HTTP_FORBIDDEN  # права пересчитаны на следующем запросе
+
+
+@pytest.mark.integration
+async def test_last_admin_cannot_deactivate_self(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
+    """Review Focus 5: последний активный с `USERS_ADMIN` не может деактивировать себя; при втором — может."""
+    admin_id = await _user_id(seeded, ADMIN)
+    admin = await auth_client(ADMIN)
+    response = await admin.patch(f"{USERS}/{admin_id}", json={"is_active": False, "full_name": "Не применится"})
+    assert response.status_code == HTTP_CONFLICT
+    assert _detail_code(response) == "LAST_ADMIN"
+    user = await seeded.get(AppUser, admin_id)
+    assert user is not None
+    await seeded.refresh(user)
+    assert user.is_active is True and user.full_name != "Не применится"
+    assert await _events(seeded, AuditAction.USER_CHANGE) == []
+
+    await make_user(
+        seeded, email="second.admin@example.com", office=await _office(seeded, HEAD_OFFICE), roles=["ADMIN"]
+    )
+    allowed = await admin.patch(f"{USERS}/{admin_id}", json={"is_active": False})
+    assert allowed.status_code == HTTP_OK, allowed.text
+    assert allowed.json()["is_active"] is False
+    assert (await admin.get(USERS)).status_code == HTTP_FORBIDDEN  # отключённый пользователь — 403
+
+
+@pytest.mark.integration
+async def test_deactivating_other_admin_respects_last_admin(
+    auth_client: AuthClientFactory, seeded: AsyncSession
+) -> None:
+    second = await make_user(
+        seeded, email="second.admin@example.com", office=await _office(seeded, HEAD_OFFICE), roles=["ADMIN"]
+    )
+    admin = await auth_client(ADMIN)
+    second_client = await auth_client("second.admin@example.com")
+    assert (await admin.patch(f"{USERS}/{second.id}", json={"is_active": False})).status_code == HTTP_OK
+    # второй отключён — первый снова последний
+    admin_id = await _user_id(seeded, ADMIN)
+    response = await admin.patch(f"{USERS}/{admin_id}", json={"is_active": False})
+    assert response.status_code == HTTP_CONFLICT
+    assert (await second_client.get(USERS)).status_code == HTTP_FORBIDDEN
+
+
+# --- Матрица прав ------------------------------------------------------------------------------------------------
+
+
+@pytest.mark.integration
+async def test_roles_and_permissions_listing(auth_client: AuthClientFactory) -> None:
+    admin = await auth_client(ADMIN)
+    permissions = (await admin.get("/admin/permissions")).json()
+    assert len(permissions) == PERMISSIONS_TOTAL
+    codes = [item["code"] for item in permissions]
+    assert codes == sorted(codes) and "USERS_ADMIN" in codes
+    users_admin = next(item for item in permissions if item["code"] == "USERS_ADMIN")
+    assert set(users_admin) == {"code", "object_type", "action", "name", "description"}
+    assert (users_admin["object_type"], users_admin["action"]) == ("USERS", "ADMIN")
+
+    roles = {item["code"]: item for item in (await admin.get("/admin/roles")).json()}
+    assert {"ADMIN", "EXPERT", "KA_RF", "KA_GO"} <= set(roles)
+    assert roles["EXPERT"]["permissions"] == EXPERT_PERMISSIONS
+    assert "USERS_ADMIN" in roles["ADMIN"]["permissions"]
+    assert set(roles["ADMIN"]) == {"code", "name", "description", "is_system", "is_active", "permissions"}
+    assert roles["ADMIN"]["is_system"] is True
+
+
+@pytest.mark.integration
+async def test_role_permissions_change_reaches_me_and_audit(
+    auth_client: AuthClientFactory, seeded: AsyncSession
+) -> None:
+    admin = await auth_client(ADMIN)
+    expert = await auth_client("expert@armka.local")
+    assert "EXPORT" not in (await expert.get("/me")).json()["permissions"]
+
+    response = await admin.put("/admin/roles/EXPERT/permissions", json={"permissions": [*EXPERT_PERMISSIONS, "EXPORT"]})
+    assert response.status_code == HTTP_OK, response.text
+    assert response.json()["permissions"] == sorted([*EXPERT_PERMISSIONS, "EXPORT"])
+
+    role = (await seeded.execute(select(Role).where(Role.code == "EXPERT"))).scalar_one()
+    row = (
+        await seeded.execute(
+            select(RolePermission)
+            .join(Permission, Permission.id == RolePermission.permission_id)
+            .where(RolePermission.role_id == role.id, Permission.code == "EXPORT")
+        )
+    ).scalar_one()
+    admin_id = await _user_id(seeded, ADMIN)
+    assert row.granted_by == admin_id and row.granted_at is not None
+    assert "EXPORT" in (await expert.get("/me")).json()["permissions"]
+
+    events = await _events(seeded, AuditAction.ROLE_PERMISSION_CHANGE)
+    assert len(events) == 1
+    event = events[0]
+    assert (event.section, event.object_type, event.object_id, event.user_id) == (
+        AuditSection.ACCESS,
+        "role",
+        role.id,
+        admin_id,
+    )
+    assert event.before_value == {"permissions": EXPERT_PERMISSIONS}
+    assert event.after_value == {"permissions": sorted([*EXPERT_PERMISSIONS, "EXPORT"])}
+
+    same = await admin.put("/admin/roles/EXPERT/permissions", json={"permissions": [*EXPERT_PERMISSIONS, "EXPORT"]})
+    assert same.status_code == HTTP_OK
+    assert len(await _events(seeded, AuditAction.ROLE_PERMISSION_CHANGE)) == 1
+
+    shrink = await admin.put("/admin/roles/EXPERT/permissions", json={"permissions": ["DIRECTORY_VIEW"]})
+    assert shrink.status_code == HTTP_OK
+    assert (await expert.get("/me")).json()["permissions"] == ["DIRECTORY_VIEW"]
+
+
+@pytest.mark.integration
+async def test_role_permissions_validation_and_last_admin(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
+    admin = await auth_client(ADMIN)
+    roles = {item["code"]: item for item in (await admin.get("/admin/roles")).json()}
+    without_users_admin = [code for code in roles["ADMIN"]["permissions"] if code != "USERS_ADMIN"]
+
+    last = await admin.put("/admin/roles/ADMIN/permissions", json={"permissions": without_users_admin})
+    assert last.status_code == HTTP_CONFLICT
+    assert _detail_code(last) == "LAST_ADMIN"
+
+    unknown = await admin.put("/admin/roles/EXPERT/permissions", json={"permissions": ["KFO_VIEW", "FLY"]})
+    assert unknown.status_code == HTTP_UNPROCESSABLE
+    assert unknown.json()["detail"]["code"] == "UNKNOWN_PERMISSION"
+    assert unknown.json()["detail"]["permissions"] == ["FLY"]
+    missing = await admin.put("/admin/roles/NO_SUCH/permissions", json={"permissions": []})
+    assert missing.status_code == HTTP_NOT_FOUND
+    assert await _events(seeded, AuditAction.ROLE_PERMISSION_CHANGE) == []
+
+    # USERS_ADMIN есть у другой роли активного пользователя — снять с ADMIN можно
+    await admin.put("/admin/roles/EXPERT/permissions", json={"permissions": [*EXPERT_PERMISSIONS, "USERS_ADMIN"]})
+    allowed = await admin.put("/admin/roles/ADMIN/permissions", json={"permissions": without_users_admin})
+    assert allowed.status_code == HTTP_OK, allowed.text
+    assert "USERS_ADMIN" not in allowed.json()["permissions"]
+
+
+# --- Права доступа -----------------------------------------------------------------------------------------------
+
+
+@pytest.mark.integration
+async def test_admin_routes_forbidden_without_users_admin(auth_client: AuthClientFactory, seeded: AsyncSession) -> None:
+    client = await auth_client("ka_go@armka.local")
+    user_id = await _user_id(seeded, "ka_rf@armka.local")
+    calls = [
+        ("GET", USERS, None),
+        ("POST", USERS, {"email": "x@example.com", "full_name": "Х", "regional_office_id": str(uuid.uuid4())}),
+        ("PATCH", f"{USERS}/{user_id}", {"full_name": "Х"}),
+        ("PATCH", f"{USERS}/{uuid.uuid4()}", {"full_name": "Х"}),
+        ("PUT", f"{USERS}/{user_id}/roles", {"roles": ["ADMIN"]}),
+        ("GET", "/admin/roles", None),
+        ("GET", "/admin/permissions", None),
+        ("PUT", "/admin/roles/KA_GO/permissions", {"permissions": ["USERS_ADMIN"]}),
+        ("PUT", "/admin/roles/NO_SUCH/permissions", {"permissions": []}),
+        ("PATCH", f"{USERS}/not-a-uuid", {"bad": 1}),
+    ]
+    for method, path, body in calls:
+        response = await client.request(method, path, json=body)
+        assert response.status_code == HTTP_FORBIDDEN, (method, path, response.text)
+        assert response.json() == {"detail": "недостаточно прав: требуется USERS_ADMIN"}
+    assert await _role_codes(seeded, user_id) == ["KA_RF"]
+
+
+# --- Гонка двух администраторов ----------------------------------------------------------------------------------
+
+
+async def _remove_admin_role(maker: async_sessionmaker[AsyncSession], actor_id: uuid.UUID, target_id: uuid.UUID) -> str:
+    """Отдельная транзакция «как запрос»: `actor_id` заменяет роли `target_id` на `[EXPERT]`; итог — `ok` или код."""
+    try:
+        async with maker() as session, session.begin():
+            audit = RequestAudit(service=AuditService(session), user_id=actor_id, ip_address=None, user_agent=None)
+            await UsersAdminService(session, audit, actor_id).set_roles(target_id, ["EXPERT"])
+    except DomainError as error:
+        return error.code
+    return "ok"
+
+
+@pytest.mark.integration
+async def test_concurrent_mutual_admin_removal_keeps_one_admin(
+    committed_sessionmaker: async_sessionmaker[AsyncSession], monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Два администратора в параллельных транзакциях снимают ADMIN друг с друга: успевает ровно один (D-8).
+
+    Без блокировки обе транзакции насчитали бы «другого» администратора и обе зафиксировались бы — ноль администраторов.
+    Проверка на уровне сервиса: обе транзакции стартуют одновременно, без предварительной аутентификации запроса.
+    Барьер перед `lock_permission` гарантирует, что обе транзакции уже прочитали своих пользователей и дошли до проверки
+    до того, как любая из них зафиксируется (без барьера первая могла бы успеть целиком и гонки бы не было).
+    """
+    barrier = asyncio.Barrier(2)
+    original_lock = users_repo.lock_permission
+
+    async def lock_after_barrier(session: AsyncSession, code: str) -> None:
+        async with asyncio.timeout(10):
+            await barrier.wait()
+        await original_lock(session, code)
+
+    monkeypatch.setattr(users_repo, "lock_permission", lock_after_barrier)
+
+    async with committed_sessionmaker() as session, session.begin():
+        first_id = await _user_id(session, ADMIN)
+        second = await make_user(
+            session, email="second.admin@example.com", office=await _office(session, HEAD_OFFICE), roles=["ADMIN"]
+        )
+        second_id = second.id
+
+    results = await asyncio.gather(
+        _remove_admin_role(committed_sessionmaker, first_id, second_id),
+        _remove_admin_role(committed_sessionmaker, second_id, first_id),
+    )
+    assert sorted(results) == ["LAST_ADMIN", "ok"]
+
+    async with committed_sessionmaker() as session:
+        remaining = await users_repo.count_active_users_with_permission(session, "USERS_ADMIN")
+    assert remaining == 1

docs/superpowers/plans/2026-10-04-slice-1-directories-phase2.md

@@ -196,9 +196,9 @@ async def run(self, *, filename: str, content: bytes, dry_run: bool, on_existing
 - Схемы (`schemas/admin.py`): `UserCreate(email: EmailStr, full_name: str, regional_office_id: uuid.UUID, employee_no: str | None = None, business_segment: BusinessSegment | None = None, roles: list[str] = [])`, `UserUpdate` (все optional, `extra="forbid"`), `UserOut(id, email, full_name, employee_no, regional_office: RegionalOfficeOut, business_segment, is_active, roles: list[str], created_at, updated_at)`, `RolesSet(roles: list[str])`, `RoleOut(code, name, description, is_system, is_active, permissions: list[str])`, `PermissionOut(code, object_type, action, name, description)`, `PermissionsSet(permissions: list[str])`.
 - Маршруты (`require("USERS_ADMIN")`): `GET /admin/users` → `Page[UserOut]` (`q`, `is_active`), `POST /admin/users` → 201, `PATCH /admin/users/{id}`, `PUT /admin/users/{id}/roles` → `UserOut`, `GET /admin/roles` → `list[RoleOut]`, `GET /admin/permissions` → `list[PermissionOut]`, `PUT /admin/roles/{code}/permissions` → `RoleOut`.
 
-- [ ] **Step 1: Тесты** (integration, `admin@armka.local`): создание с `roles=["KA_RF"]` → `app_user`+`user_role`; `GET /me` новым пользователем (через `auth_client(new_email)`) → права KA_RF; email в другом регистре → 409 `EMAIL_EXISTS`; `PUT roles` `[KA_GO, EXPERT]` → ровно две строки, аудит before/after; единственный админ снимает с себя `ADMIN` → 409 `LAST_ADMIN`; второй админ создан → снятие разрешено; `PATCH is_active=false` на себе как последнем → 409; матрица: `EXPERT` + `EXPORT` → `role_permission` строка, `/me` экспертом видит `EXPORT`, аудит `ROLE_PERMISSION_CHANGE`; убрать `USERS_ADMIN` у `ADMIN` → 409 `LAST_ADMIN`; `GET /admin/permissions` → 16 кодов; `ka_go` → 403 на всё.
-- [ ] **Step 2–4:** падает → реализация → тесты, `make check` → PASS.
-- [ ] **Step 5: Commit** `feat(admin): пользователи, роли, матрица прав — API с аудитом и защитой LAST_ADMIN`.
+- [x] **Step 1: Тесты** (integration, `admin@armka.local`): создание с `roles=["KA_RF"]` → `app_user`+`user_role`; `GET /me` новым пользователем (через `auth_client(new_email)`) → права KA_RF; email в другом регистре → 409 `EMAIL_EXISTS`; `PUT roles` `[KA_GO, EXPERT]` → ровно две строки, аудит before/after; единственный админ снимает с себя `ADMIN` → 409 `LAST_ADMIN`; второй админ создан → снятие разрешено; `PATCH is_active=false` на себе как последнем → 409; матрица: `EXPERT` + `EXPORT` → `role_permission` строка, `/me` экспертом видит `EXPORT`, аудит `ROLE_PERMISSION_CHANGE`; убрать `USERS_ADMIN` у `ADMIN` → 409 `LAST_ADMIN`; `GET /admin/permissions` → 16 кодов; `ka_go` → 403 на всё.
+- [x] **Step 2–4:** падает → реализация → тесты, `make check` → PASS.
+- [x] **Step 5: Commit** `feat(admin): пользователи, роли, матрица прав — API с аудитом и защитой LAST_ADMIN`.
 
 ---
 

openspec/changes/slice-1-directories/tasks.md

@@ -63,7 +63,7 @@ test-integration`). Пути — от корня репозитория. Доп
 
 ## 4. Администрирование
 
-- [ ] 4.1 **Пользователи и роли (API).** `services/users_admin.py` (создание с проверкой `lower(email)`, изменение, деактивация,
+- [x] 4.1 **Пользователи и роли (API).** `services/users_admin.py` (создание с проверкой `lower(email)`, изменение, деактивация,
       `PUT roles` с diff и `LAST_ADMIN`), `repositories/users.py` (+ запросы списка), эндпоинты `api/endpoints/admin/users.py`,
       `roles.py` (`GET roles`, `GET permissions`, `PUT roles/{code}/permissions` с `LAST_ADMIN` и защитой `is_system`),
       аудит `USER_CHANGE` / `USER_ROLE_CHANGE` / `ROLE_PERMISSION_CHANGE`. Тесты: сценарии спеки `users-admin` (создание с ролью →