gendesign/tradein-mvp/backend/app/api/v1/team.py

"""Team-management API — CRUD сотрудников, квоты, история (#2554, эпик #2549).

Mounted at `/api/v1/team`; через Caddy `uri strip_prefix /trade-in` это
`/trade-in/api/v1/team/*` снаружи. `app.services.auth_session.DB_ROLE_PATHS`
уже закладывает `/api/v1/team/**` в scope роли `manager` (и `admin` через `/**`)
для `rbac_guard` (см. `app.core.rbac`) — этот роутер добавляет ВТОРОЙ,
более узкий барьер именно на identity:

  - `current_team_actor` резолвит юзера ТОЛЬКО из session-cookie
    (`app.services.auth_session.get_session_user`). Legacy
    `X-Authenticated-User` (Caddy trusted-header, dual-mode) НЕ принимается
    здесь — team-API новый, не участвует в переходном dual-mode auth. Без
    валидной cookie — 401, даже если `rbac_guard` пропустил запрос по
    legacy-заголовку (напр. admin через roles.yaml).
  - Роль должна быть `admin` или `manager` — иначе 403.

Org-изоляция (главный инвариант фичи): manager видит/меняет ТОЛЬКО своих
employee (`<реестр>.manager_id = actor.user_id`). Чужой/несуществующий
employee_id → 404 (НЕ 403) — не подтверждаем/не опровергаем существование
чужого сотрудника перед manager'ом. См. `_authorize_employee`.

ДВЕ СЕССИИ БД, и это не дублирование:
  - `identity_db` (`Depends(get_identity_db)`) — реестр людей: строка сотрудника
    и его сессии. При `IDENTITY_STORE=auth` это ДРУГАЯ БД (`auth`).
  - `db` (`Depends(get_db)`) — продуктовые таблицы «Меры», которые в общий
    реестр не переезжают: `account_quota_overrides`, `account_estimate_usage`,
    `user_events`, `trade_in_estimates`.
В дефолтном режиме (`IDENTITY_STORE=tradein`) это ОДИН И ТОТ ЖЕ объект `Session`
(см. `identity_store.get_identity_db`), поэтому всё по-прежнему коммитится одной
транзакцией — прод не меняется. В режиме `auth` транзакции физически две:
порядок коммитов выбран так, чтобы при сбое второго коммита оставалось менее
вредное состояние (см. комментарии у `db.commit()`), а `db is not identity_db` —
рантайм-признак «БД разные».

Гранты роли `auth_app` (data/sql/auth/004, Часть 4) этот роутер соблюдает без
обходов: он ПИШЕТ только `password_hash, display_name, org_name, email,
access_state, updated_at` (ровно column-level GRANT UPDATE), вставляет строку
целиком (табличный GRANT INSERT) и НИКОГДА не пишет `role`/`manager_id`
UPDATE'ом и не делает DELETE по `users`.

DELETE по `sessions` реестра — штатный и грантом предусмотрен (data/sql/auth/002,
GRANT DELETE на sessions): блокировка и смена пароля обязаны рвать живые сессии
немедленно, это `revoke_user_sessions` из `app.services.auth_session`, вызываемый
из `update_employee`. То есть периметр DELETE у этого роутера — ровно `sessions`
и ничего больше; грант DELETE на sessions не лишний.

Кого именно можно менять через этот роутер (`_MANAGEABLE_ROLES_BY_ACTOR`):
  - actor manager → только `role='employee'` И только своих (как было).
  - actor admin   → `role IN ('employee','manager')`.

Почему admin'у отдали и менеджеров (инцидент 2026-07-31): после cutover'а на
DB-auth (#2558) аккаунты `kopylov`/`praktika` сидят с `role='manager'`, а этот
роутер жёстко фильтровал `role='employee'` — сбросить менеджеру пароль или
заблокировать его было НЕЧЕМ, кроме ручного psql на проде. Роль manager вводилась
как «владелец своей организации», а не как «неприкасаемый аккаунт».

`role='admin'` НЕ входит ни в один набор, и это несущий инвариант, а не
экономия: он один держит невозможность self-lockout'а. Актёр этого роутера —
всегда admin или manager (`current_team_actor`); manager до admin-строки не
дотянется по своей ветке фильтра, а admin не дотянется до admin-строки вообще —
в том числе до собственной. Поэтому ни один путь ниже (block, смена пароля +
`revoke_user_sessions`) не может вырубить самого действующего админа или
разжаловать другого. Раздача/отзыв роли admin остаётся операцией уровня
миграции/psql — сознательно вне API.
"""

from __future__ import annotations

import logging
from dataclasses import dataclass
from typing import Annotated, Any
from urllib.parse import urlparse

from fastapi import APIRouter, Depends, HTTPException, Query, Request
from sqlalchemy import text
from sqlalchemy.engine import RowMapping
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session
from sqlalchemy.sql.elements import TextClause

from app.core.auth import get_role
from app.core.config import settings
from app.core.db import get_db
from app.core.password import hash_password
from app.schemas.team import (
    EmployeeCreateRequest,
    EmployeeHistoryEntry,
    EmployeeOut,
    EmployeeUpdateRequest,
    QuotaStatusOut,
)
from app.services import account_quota
from app.services.auth_session import get_session_user, revoke_user_sessions
from app.services.identity_store import (
    AccessState,
    IdentitySchema,
    access_state_param,
    get_identity_db,
    identity_schema,
    to_access_state,
)
from app.services.user_events import schedule_event

logger = logging.getLogger(__name__)

router = APIRouter()


@dataclass
class TeamActor:
    """Резолвленный из session-cookie актёр team-API — admin или manager."""

    user_id: int
    username: str
    role: str  # "admin" | "manager"


async def current_team_actor(
    request: Request,
    identity_db: Annotated[Session, Depends(get_identity_db)],
) -> TeamActor:
    """Dependency: session-only identity, роль admin|manager, иначе 401/403.

    Намеренно НЕ читает `X-Authenticated-User` — см. модульный docstring.
    Сессия резолвится в БД РЕЕСТРА (см. про две сессии в модульном docstring).
    """
    token = request.cookies.get(settings.session_cookie_name)
    if not token:
        raise HTTPException(status_code=401, detail="valid session required")

    try:
        session_user = get_session_user(identity_db, token)
    except Exception:
        logger.exception("team: session lookup failed")
        raise HTTPException(status_code=401, detail="valid session required") from None

    if session_user is None:
        raise HTTPException(status_code=401, detail="valid session required")

    role = session_user["role"]
    if role not in ("admin", "manager"):
        raise HTTPException(status_code=403, detail="admin or manager role required")

    return TeamActor(
        user_id=session_user["user_id"],
        username=session_user["username"],
        role=role,
    )


def _origin_host_allowed(candidate: str) -> bool:
    """True если scheme://netloc *candidate* совпадает с одним из `settings.cors_origins`.

    `cors_origins` уже является источником правды для «какие origin'ы это наш
    фронт» (см. CORSMiddleware в app/main.py, ENV CORS_ORIGINS) — переиспользуем
    его вместо нового хардкода."""
    try:
        parsed = urlparse(candidate)
    except ValueError:
        return False
    if not parsed.scheme or not parsed.netloc:
        return False
    origin = f"{parsed.scheme}://{parsed.netloc}"
    return origin in settings.cors_origins


def _require_same_origin(request: Request) -> None:
    """CSRF defense-in-depth (issue #2554 DoD) для state-changing team-роутов
    (POST/PATCH): `Origin` (или `Referer` как fallback) обязан матчить один из
    `settings.cors_origins`, иначе 403.

    Оба заголовка отсутствуют → ПРОПУСКАЕМ (не 403). Причина: это единственный
    надёжный сигнал non-browser клиента в этом стеке — curl-смоуки внутри
    контейнера (см. `.claude/rules/tradein.md` "Тестировать HTTP только ВНУТРИ
    контейнера", `docker exec tradein-backend curl ...`) не шлют ни один из этих
    заголовков, а реальный браузер (fetch/XHR/form) ВСЕГДА прикладывает Origin
    на unsafe-методах (POST/PATCH) — так что "оба отсутствуют" практически
    невозможно для настоящего кросс-сайтового CSRF через браузер. Session-cookie
    уже стоит на `SameSite=Lax` (см. `app.api.v1.auth.login`) — это первый рубеж
    против CSRF, Origin-check — второй.
    """
    candidate = request.headers.get("origin") or request.headers.get("referer")
    if candidate is None:
        return
    if not _origin_host_allowed(candidate):
        logger.warning(
            "team: Origin/Referer mismatch %r on %s — possible CSRF", candidate, request.url.path
        )
        raise HTTPException(status_code=403, detail="origin not allowed")


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


# Имена таблицы и колонки состояния доступа приходят из `identity_schema()` —
# фиксированный словарь в `app.services.identity_store`, единственный источник
# этих имён (в SQL-строку не попадает ничего пришедшего снаружи; значения
# по-прежнему биндятся параметрами).
#
# `AS access_state` в КАЖДОМ SELECT'е — не косметика: колонка называется
# по-разному в двух схемах, и без алиаса вызывающий код читал бы то `is_active`,
# то `access_state`, то есть завёл бы то самое второе представление состояния,
# которого быть не должно. Дальше значение всегда идёт через `to_access_state()`.
def _employee_columns(schema: IdentitySchema) -> str:
    return (
        "id, username, role, display_name, org_name, email, "
        f"{schema.access_state_column} AS access_state, manager_id, created_at"
    )


# Два статических варианта — НЕ динамическая сборка WHERE (та же мотивация, что
# у `_list_employees_sql` ниже: значения и так биндятся параметрами, но
# статические ветки не провоцируют будущие правки в сторону конкатенации SQL).
# Роль 'admin' не встречается ни в одной ветке — см. модульный docstring.
def _fetch_employee_sql(actor_role: str) -> TextClause:
    schema = identity_schema()
    cols = _employee_columns(schema)
    if actor_role == "admin":
        return text(
            f"SELECT {cols} FROM {schema.users_table} "
            "WHERE id = :id AND role IN ('employee', 'manager')"
        )
    return text(f"SELECT {cols} FROM {schema.users_table} WHERE id = :id AND role = 'employee'")


def _fetch_employee_row(
    identity_db: Session, employee_id: int, actor: TeamActor
) -> RowMapping | None:
    """Строка управляемого юзера в пределах прав *actor* — иначе None (→ 404).

    Фильтр по роли делается ЗДЕСЬ, в SQL, а не в `_authorize_employee` ниже:
    для manager'а строка менеджера/админа не должна даже доехать до
    вызывающего кода. `None` для обоих случаев («нет такого id» и «этот id
    тебе не по зубам») — тот же принцип, что и 404-вместо-403 в
    `_authorize_employee`: не палим существование чужой строки.
    """
    sql = _fetch_employee_sql(actor.role)
    return identity_db.execute(sql, {"id": employee_id}).mappings().fetchone()


def _authorize_employee(actor: TeamActor, row: RowMapping | None) -> RowMapping:
    """404 (НЕ 403) если сотрудник не найден ИЛИ принадлежит другому manager'у.

    Org-изоляция: manager может видеть/менять только `manager_id == actor.user_id`.
    404 вместо 403 — не палим существование чужого employee_id.

    Для admin'а доп. проверки нет: набор строк, до которых он вообще может
    дотянуться, уже ограничен ролью в `_fetch_employee_row` (employee|manager,
    без admin). У менеджерских строк `manager_id` штатно NULL — сравнивать его
    с чем-либо здесь нечего.
    """
    if row is None:
        raise HTTPException(status_code=404, detail="employee not found")
    if actor.role == "manager" and row["manager_id"] != actor.user_id:
        raise HTTPException(status_code=404, detail="employee not found")
    return row


def _upsert_quota_override(
    db: Session, username: str, monthly_limit: int, actor_username: str
) -> None:
    """Upsert персонального лимита. Явная установка monthly_limit — сигнал "хочу
    numeric-квоту", поэтому ВСЕГДА сбрасывает `unlimited=false` (иначе лимит может
    молча не применяться — прежний unlimited-грант выигрывал бы у нового limit).
    `note` — НЕ затирается, если уже задан (`COALESCE`): не перезаписываем
    человеко-читаемую причину прошлого гранта (напр. "пилот, грант ...") молча
    сгенерированной строкой; note проставляется только при первом upsert записи.
    """
    db.execute(
        text(
            """
            INSERT INTO account_quota_overrides (username, monthly_limit, unlimited, note)
            VALUES (:username, CAST(:monthly_limit AS integer), false, :note)
            ON CONFLICT (username) DO UPDATE SET
                monthly_limit = EXCLUDED.monthly_limit,
                unlimited = false,
                note = COALESCE(account_quota_overrides.note, EXCLUDED.note),
                updated_at = now()
            """
        ),
        {
            "username": username,
            "monthly_limit": monthly_limit,
            "note": f"team-api: set by {actor_username}",
        },
    )


def _batch_quota_status(db: Session, usernames: list[str]) -> dict[str, dict[str, Any]]:
    """Батч-версия `account_quota.get_status` для N сотрудников — 2 SQL-запроса
    вместо 2N (было 2N+3 на GET /employees, HIGH/Medium2 review PR #2563).

    Семантика ИДЕНТИЧНА `account_quota.is_unlimited`/`user_limit`/`get_status`
    (follow-up review PR #2563 п.2 — предыдущая версия расходилась: батч ВСЕГДА
    читал `account_quota_overrides.unlimited`, а `is_unlimited` — ТОЛЬКО для
    username, присутствующего в roles.yaml):
      - username НЕ в roles.yaml (`get_role` → KeyError) → unlimited=False ВСЕГДА,
        `account_quota_overrides.unlimited` даже не проверяется (roles.yaml —
        источник правды "кто вообще может быть unlimited", override — "у кого
        именно из известных roles.yaml-юзеров"). Сегодня недостижимо для DB-only
        сотрудников team-API (`_upsert_quota_override` всегда пишет
        `unlimited=false`), но станет достижимым при ручном UPDATE
        `account_quota_overrides` или расширении roles.yaml — расхождение с
        реальным enforcement (`check_and_raise`/`increment`, тот же `is_unlimited`)
        было бы честной ложью в списке: "без лимита", который движок всё равно
        считает.
      - username в roles.yaml и role == admin → unlimited=True (без похода в БД).
      - username в roles.yaml, role != admin → unlimited = override.unlimited.
    limit = override.monthly_limit (читается для ЛЮБОГО username, без gate по
    roles.yaml — так же ведёт себя `account_quota.user_limit`), иначе глобальный
    `account_quota.MONTHLY_LIMIT`.
    """
    if not usernames:
        return {}

    overrides = (
        db.execute(
            text(
                """
                SELECT username, monthly_limit, unlimited
                FROM account_quota_overrides
                WHERE username = ANY(CAST(:usernames AS text[]))
                """
            ),
            {"usernames": usernames},
        )
        .mappings()
        .all()
    )
    override_by_username = {r["username"]: r for r in overrides}

    period = account_quota.current_period()
    usage_rows = (
        db.execute(
            text(
                """
                SELECT username, used
                FROM account_estimate_usage
                WHERE username = ANY(CAST(:usernames AS text[])) AND period_month = :period
                """
            ),
            {"usernames": usernames, "period": period},
        )
        .mappings()
        .all()
    )
    used_by_username = {r["username"]: r["used"] for r in usage_rows}

    result: dict[str, dict[str, Any]] = {}
    for username in usernames:
        override = override_by_username.get(username)
        try:
            role = get_role(username)
        except KeyError:
            role = None
        if role == "admin":
            unlimited = True
        elif role is not None:
            unlimited = bool(override is not None and override["unlimited"])
        else:
            # username не в roles.yaml — is_unlimited() короткое замыкание на
            # False, override НЕ проверяется (см. докстринг выше).
            unlimited = False
        limit = (
            int(override["monthly_limit"])
            if override is not None and override["monthly_limit"] is not None
            else account_quota.MONTHLY_LIMIT
        )
        used = used_by_username.get(username, 0)
        if unlimited:
            result[username] = {
                "limit": limit,
                "used": used,
                "remaining": limit,
                "unlimited": True,
            }
        else:
            remaining = max(0, limit - max(0, used))
            result[username] = {
                "limit": limit,
                "used": used,
                "remaining": remaining,
                "unlimited": False,
            }
    return result


def _employee_out(row: RowMapping, quota: dict[str, Any]) -> EmployeeOut:
    """Строка реестра → ответ API.

    `is_active` в контракте API остаётся булевым (форма ответа не меняется —
    фронт «Команды» не трогаем этим PR), и считается он ровно как «пустят ли
    входить»: `trial_expired` показывается как заблокированный. Отдельное
    отображение пробного периода в «Команде» — вопрос UI-PR'а, не этого.
    """
    return EmployeeOut(
        id=row["id"],
        username=row["username"],
        role=row["role"],
        display_name=row["display_name"],
        org_name=row["org_name"],
        email=row["email"],
        is_active=to_access_state(row["access_state"]).can_sign_in,
        manager_id=row["manager_id"],
        created_at=row["created_at"],
        quota=QuotaStatusOut(**quota),
    )


# ---------------------------------------------------------------------------
# POST /employees
# ---------------------------------------------------------------------------


@router.post("/employees", response_model=EmployeeOut, status_code=201)
async def create_employee(
    body: EmployeeCreateRequest,
    actor: Annotated[TeamActor, Depends(current_team_actor)],
    db: Annotated[Session, Depends(get_db)],
    identity_db: Annotated[Session, Depends(get_identity_db)],
    _origin_check: Annotated[None, Depends(_require_same_origin)],
) -> EmployeeOut:
    """Создать сотрудника. Роль всегда `employee`.

    manager_id: для actor.role == manager — принудительно свой id (любое
    значение из тела ИГНОРИРУЕТСЯ, org-изоляция инвариант #2554). Для
    actor.role == admin — опционально из тела, валидируется что указанный id
    существует и role='manager' (иначе 422).

    `identity_db` — реестр (строка сотрудника), `db` — продуктовая квота;
    в дефолтном режиме это одна и та же сессия и одна транзакция.
    """
    schema = identity_schema()
    existing = identity_db.execute(
        text(f"SELECT id FROM {schema.users_table} WHERE username = :u"),
        {"u": body.username},
    ).fetchone()
    if existing is not None:
        raise HTTPException(status_code=409, detail="username already exists")

    try:
        password_hash = hash_password(body.password)
    except ValueError as e:
        raise HTTPException(status_code=422, detail=str(e)) from None

    manager_id: int | None
    if actor.role == "manager":
        # Инвариант org-изоляции: manager не может создать сотрудника под
        # чужим manager_id — любое значение из тела игнорируется молча.
        manager_id = actor.user_id
    else:
        manager_id = body.manager_id
        if manager_id is not None:
            mgr = identity_db.execute(
                text(f"SELECT id FROM {schema.users_table} WHERE id = :id AND role = 'manager'"),
                {"id": manager_id},
            ).fetchone()
            if mgr is None:
                raise HTTPException(
                    status_code=422,
                    detail="manager_id does not reference an existing manager",
                )

    try:
        row = (
            identity_db.execute(
                text(
                    f"""
                INSERT INTO {schema.users_table}
                    (username, password_hash, role, manager_id, display_name, org_name,
                     email, {schema.access_state_column})
                VALUES
                    (:username, :password_hash, 'employee', :manager_id, :display_name,
                     :org_name, :email, :access_state)
                RETURNING {_employee_columns(schema)}
                """
                ),
                {
                    "username": body.username,
                    "password_hash": password_hash,
                    "manager_id": manager_id,
                    "display_name": body.display_name,
                    "org_name": body.org_name,
                    "email": body.email,
                    # Новый сотрудник заводится с открытым доступом — как и
                    # раньше (`is_active = true` литералом). Литерала здесь
                    # больше нет: тип колонки разный, знает о нём identity_store.
                    "access_state": access_state_param(AccessState.ACTIVE),
                },
            )
            .mappings()
            .fetchone()
        )
    except IntegrityError:
        # TOCTOU: два конкурентных POST с одинаковым username между pre-check
        # выше и этим INSERT — UNIQUE-констрейнт на username в реестре ловит.
        identity_db.rollback()
        raise HTTPException(status_code=409, detail="username already exists") from None

    assert row is not None  # RETURNING на успешный INSERT всегда отдаёт строку

    if body.monthly_limit is not None:
        _upsert_quota_override(db, body.username, body.monthly_limit, actor.username)

    # Реестр коммитится ПЕРВЫМ. В дефолтном режиме это один коммит на одну
    # транзакцию (identity_db is db) — ровно как было. В режиме `auth` БД две,
    # и порядок выбран по цене сбоя: не доехавшая квота — это сотрудник с
    # глобальным лимитом (чинится повторным PATCH), тогда как не доехавшая
    # строка сотрудника при уже сохранённой квоте — висящий override на
    # несуществующего человека.
    identity_db.commit()
    if db is not identity_db:
        db.commit()

    schedule_event(
        event_type="employee_created",
        username=actor.username,
        payload={
            "employee_id": row["id"],
            "employee_username": row["username"],
            "manager_id": manager_id,
        },
    )

    quota = account_quota.get_status(db, body.username)
    return _employee_out(row, quota)


# ---------------------------------------------------------------------------
# PATCH /employees/{id}
# ---------------------------------------------------------------------------


@router.patch("/employees/{employee_id}", response_model=EmployeeOut)
async def update_employee(
    employee_id: int,
    body: EmployeeUpdateRequest,
    actor: Annotated[TeamActor, Depends(current_team_actor)],
    db: Annotated[Session, Depends(get_db)],
    identity_db: Annotated[Session, Depends(get_identity_db)],
    _origin_check: Annotated[None, Depends(_require_same_origin)],
) -> EmployeeOut:
    """Частичное обновление сотрудника — block/unblock, лимит, профиль, пароль.

    manager может патчить ТОЛЬКО своих (manager_id == actor.user_id), иначе 404.
    При is_active=False ИЛИ смене пароля (new_password) — обязательно revoke всех
    сессий (HIGH, deep-review PR #2563): без этого блокировка/reset не подействуют
    до истечения TTL текущей сессии сотрудника — хуже того, sliding-refresh
    (`app.services.auth_session.get_session_user`) продлевает `expires_at` на
    КАЖДОМ запросе, так что скомпрометированная/чужая сессия живёт неограниченно
    долго, а не «до TTL». `revoke_user_sessions` сам называет смену пароля своим
    use-case — см. его докстринг.

    `is_active` в теле остаётся булевым (контракт API не меняется): true →
    `active`, false → `disabled`. Перевести аккаунт В `trial_expired` этим
    роутом нельзя — это состояние проставляется миграцией/владельцем, а
    выразить его булевым полем нечем; is_active=true на таком аккаунте открывает
    доступ (снимает пробное ограничение), is_active=false закрывает жёстко.
    """
    row = _fetch_employee_row(identity_db, employee_id, actor)
    row = _authorize_employee(actor, row)

    new_password_hash: str | None = None
    if body.new_password is not None:
        try:
            new_password_hash = hash_password(body.new_password)
        except ValueError as e:
            raise HTTPException(status_code=422, detail=str(e)) from None

    schema = identity_schema()
    # Пишутся РОВНО те колонки, на которые у auth_app есть column-level GRANT
    # UPDATE (data/sql/auth/004, Часть 4): password_hash, display_name, org_name,
    # email, access_state, updated_at. role и manager_id этим роутом не
    # обновляются — не «пока не понадобилось», а сознательно: право на их запись
    # роли приложения не выдано, и добавлять его в обход миграции нельзя.
    #
    # CAST обязателен из-за NULL-параметра (поле не пришло в PATCH → COALESCE
    # оставляет текущее значение): у нетипизированного NULL Postgres не может
    # вывести тип. Имя SQL-типа — из фиксированного словаря identity_store.
    identity_db.execute(
        text(
            f"""
            UPDATE {schema.users_table}
            SET display_name = COALESCE(:display_name, display_name),
                org_name = COALESCE(:org_name, org_name),
                email = COALESCE(:email, email),
                {schema.access_state_column} = COALESCE(
                    CAST(:access_state AS {schema.access_state_sql_type}),
                    {schema.access_state_column}
                ),
                password_hash = COALESCE(:password_hash, password_hash),
                updated_at = now()
            WHERE id = :id
            """
        ),
        {
            "display_name": body.display_name,
            "org_name": body.org_name,
            "email": body.email,
            "access_state": (
                None
                if body.is_active is None
                else access_state_param(
                    AccessState.ACTIVE if body.is_active else AccessState.DISABLED
                )
            ),
            "password_hash": new_password_hash,
            "id": employee_id,
        },
    )

    if body.monthly_limit is not None:
        _upsert_quota_override(db, row["username"], body.monthly_limit, actor.username)

    if body.is_active is False or body.new_password is not None:
        # Обязательно ПОСЛЕ UPDATE, ДО финального commit — revoke_user_sessions
        # коммитит сам (см. app.services.auth_session), это флашит и наш
        # предшествующий UPDATE (а в дефолтном режиме, где сессия одна, — и
        # quota-upsert). Сессии живут в БД реестра, вместе с пользователем,
        # поэтому рвём их через `identity_db`: с чужой сессией здесь блокировка
        # и смена пароля перестали бы действовать немедленно. Self-lockout
        # невозможен: _fetch_employee_row не отдаёт строки с role='admin'
        # НИКОМУ, а manager'у — ещё и только role='employee'; т.е. actor
        # (admin|manager) никогда не может патчить сам себя через этот роут.
        revoke_user_sessions(identity_db, employee_id)

    # Порядок и смысл — как в create_employee: реестр первым, продуктовая БД
    # отдельным коммитом только если она физически другая.
    identity_db.commit()
    if db is not identity_db:
        db.commit()

    changed_profile_fields = [
        f
        for f, v in (
            ("display_name", body.display_name),
            ("org_name", body.org_name),
            ("email", body.email),
        )
        if v is not None
    ]
    if changed_profile_fields:
        schedule_event(
            event_type="employee_updated",
            username=actor.username,
            payload={
                "employee_id": employee_id,
                "employee_username": row["username"],
                "fields": changed_profile_fields,
            },
        )
    if body.new_password is not None:
        schedule_event(
            event_type="employee_password_reset",
            username=actor.username,
            payload={"employee_id": employee_id, "employee_username": row["username"]},
        )
    if body.is_active is not None:
        schedule_event(
            event_type="employee_blocked" if body.is_active is False else "employee_unblocked",
            username=actor.username,
            payload={"employee_id": employee_id, "employee_username": row["username"]},
        )
    if body.monthly_limit is not None:
        schedule_event(
            event_type="quota_changed",
            username=actor.username,
            payload={
                "employee_id": employee_id,
                "employee_username": row["username"],
                "monthly_limit": body.monthly_limit,
            },
        )

    updated_row = _fetch_employee_row(identity_db, employee_id, actor)
    assert updated_row is not None  # только что успешно обновили эту же строку
    quota = account_quota.get_status(db, updated_row["username"])
    return _employee_out(updated_row, quota)


# ---------------------------------------------------------------------------
# GET /employees
# ---------------------------------------------------------------------------


# Два статических варианта WHERE (НЕ f-string/динамическая сборка — Medium/
# "заодно" review PR #2563: значения биндятся параметрами и без того безопасны,
# но статические ветки не провоцируют будущие правки в сторону конкатенации SQL).
#
# ORDER BY created_at DESC, id DESC — тай-брейкер по `id` ОБЯЗАТЕЛЕН (follow-up
# review PR #2563 п.1): `created_at DEFAULT now()` — время ТРАНЗАКЦИИ, а bulk-seed
# (#2557) вставляет много юзеров одной транзакцией → идентичный timestamp у N строк.
# Без тай-брейкера порядок между страницами (LIMIT/OFFSET) на PostgreSQL для
# строк-«близнецов» не гарантирован — сотрудники пропадали/дублировались бы при
# постраничном листании. `id` монотонно растёт (BIGINT IDENTITY) — детерминированный
# tie-break без доп. индекса (созданные позже = бОльший id, тот же порядок что и
# намерение DESC-сортировки по времени).
#
# Admin-ветка (`by_manager=False`): сюда попадают И менеджеры (см. модульный
# docstring — иначе admin не видит в UI строку, которой должен уметь сбросить
# пароль). `role='admin'` по-прежнему невидим и неуправляем. Сортировка по
# (created_at, id) общая для обеих веток — намеренно: seed (#2557) вставил всех
# одной транзакцией, так что группировка «сначала менеджеры» дала бы ложное
# ощущение иерархии там, где её в данных нет; роль показывается колонкой
# (`EmployeeOut.role`).
def _list_employees_sql(*, by_manager: bool) -> TextClause:
    schema = identity_schema()
    cols = _employee_columns(schema)
    tail = "ORDER BY created_at DESC, id DESC LIMIT :limit OFFSET :offset"
    if by_manager:
        return text(
            f"SELECT {cols} FROM {schema.users_table} "
            f"WHERE role = 'employee' AND manager_id = :manager_id {tail}"
        )
    return text(
        f"SELECT {cols} FROM {schema.users_table} WHERE role IN ('employee', 'manager') {tail}"
    )


@router.get("/employees", response_model=list[EmployeeOut])
async def list_employees(
    actor: Annotated[TeamActor, Depends(current_team_actor)],
    db: Annotated[Session, Depends(get_db)],
    identity_db: Annotated[Session, Depends(get_identity_db)],
    manager_id: Annotated[int | None, Query()] = None,
    limit: Annotated[int, Query(ge=1, le=200)] = 50,
    offset: Annotated[int, Query(ge=0)] = 0,
) -> list[EmployeeOut]:
    """Список сотрудников. manager видит только своих; admin — всех, опц. ?manager_id=.

    Сотрудники читаются из реестра (`identity_db`), квоты — из продуктовой БД
    (`db`): `account_quota_overrides`/`account_estimate_usage` в общий реестр не
    переезжают. Квота — ОДИН батч-запрос на всю страницу (`_batch_quota_status`),
    не N+1 (Medium2, review PR #2563: было 2N+3 SQL-запросов на N сотрудников).
    """
    if actor.role == "manager":
        rows = (
            identity_db.execute(
                _list_employees_sql(by_manager=True),
                {"manager_id": actor.user_id, "limit": limit, "offset": offset},
            )
            .mappings()
            .all()
        )
    elif manager_id is not None:
        rows = (
            identity_db.execute(
                _list_employees_sql(by_manager=True),
                {"manager_id": manager_id, "limit": limit, "offset": offset},
            )
            .mappings()
            .all()
        )
    else:
        rows = (
            identity_db.execute(
                _list_employees_sql(by_manager=False), {"limit": limit, "offset": offset}
            )
            .mappings()
            .all()
        )

    quota_by_username = _batch_quota_status(db, [row["username"] for row in rows])
    return [_employee_out(row, quota_by_username[row["username"]]) for row in rows]


# ---------------------------------------------------------------------------
# GET /employees/{id}/history
# ---------------------------------------------------------------------------


@router.get("/employees/{employee_id}/history", response_model=list[EmployeeHistoryEntry])
async def employee_history(
    employee_id: int,
    actor: Annotated[TeamActor, Depends(current_team_actor)],
    db: Annotated[Session, Depends(get_db)],
    identity_db: Annotated[Session, Depends(get_identity_db)],
    limit: Annotated[int, Query(ge=1, le=200)] = 50,
    offset: Annotated[int, Query(ge=0)] = 0,
) -> list[EmployeeHistoryEntry]:
    """История оценок сотрудника (адрес/дата/результат) — из `user_events`,
    LEFT JOIN `trade_in_estimates` за фактическим результатом.

    Та же org-проверка что и в PATCH: чужой employee_id → 404. Проверка идёт по
    реестру (`identity_db`), сама история — продуктовые таблицы (`db`).
    """
    row = _fetch_employee_row(identity_db, employee_id, actor)
    row = _authorize_employee(actor, row)

    rows = (
        db.execute(
            text(
                """
                SELECT
                    CAST(ue.estimate_id AS text) AS estimate_id,
                    ue.payload ->> 'address' AS address,
                    ue.payload ->> 'area_m2' AS area_m2,
                    ue.payload ->> 'rooms' AS rooms,
                    te.median_price,
                    te.confidence,
                    te.n_analogs,
                    ue.created_at
                FROM user_events ue
                LEFT JOIN trade_in_estimates te ON te.id = ue.estimate_id
                WHERE ue.username = :username AND ue.event_type = 'estimate_request'
                ORDER BY ue.created_at DESC
                LIMIT :limit OFFSET :offset
                """
            ),
            {"username": row["username"], "limit": limit, "offset": offset},
        )
        .mappings()
        .all()
    )

    return [EmployeeHistoryEntry.model_validate(dict(r)) for r in rows]