feat(domain): policy видимости справочников (§10.1, ASSUMPTION(OQ-86))

Первый модуль src/domain/ (чистые функции, без I/O, mypy strict): VisibilityScope,
scope_for/scope_of, SQL-фильтры company_groups_visible/legal_entities_visible,
проверки company_group_is_visible/legal_entity_is_visible, office_for_create
(чужой филиал → 422 OFFICE_NOT_ALLOWED). Код роли не читается; business_segment
не фильтрует (ASSUMPTION(OQ-86)). Принадлежность ЮЛ ГК филиала — подзапросом IN,
а не JOIN (число строк не меняется, нет конфликта с JOIN вызывающего).

Тесты: unit без БД (компиляция Select под PostgreSQL, все ветки) + один
integration-тест семантики фильтров и их согласованности с проверками объектов.
Отмечены шаги Task 4 в плане и 2.1 в tasks.md.

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

api/src/domain/__init__.py

@@ -0,0 +1 @@
+"""Доменный слой: чистые функции по §§ правил `05-domain-rules.md`, без I/O (ни сессий БД, ни сети, ни файлов)."""

api/src/domain/policy.py

@@ -0,0 +1,109 @@
+"""Policy видимости справочников (§10.1 `05-domain-rules.md`, spec directory-policy).
+
+Правило формулируется через данные, а не через роль (инвариант «права — данные», §10): пользователь головного
+офиса (`regional_office.is_head_office`) видит всё; пользователь филиала — ГК с `base_regional_office_id` своего
+филиала, их ЮЛ и ГСК/ГСЗ, а также монозаёмщиков своего филиала (`company_group_id IS NULL AND regional_office_id = …`).
+Код роли (`role.code`) здесь не читается. Право (`require`) отвечает «может ли вообще», policy — «какие именно
+записи»; объект вне видимости вызывающий превращает в 404 с тем же телом, что у несуществующего.
+
+ASSUMPTION(OQ-86): ЗО (`app_user.business_segment`) в Срезе 1 не фильтрует — у ГК и ЮЛ нет поля сегмента (DBML v5.14).
+"""
+
+import uuid
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Any
+
+from sqlalchemy import Select, and_, or_, select
+
+from src.common.exceptions import DomainValidationError
+from src.models.directories import CompanyGroup, LegalEntity
+
+if TYPE_CHECKING:
+    # Только для типов: домен не тянет репозитории в рантайме
+    from src.repositories.users import UserWithAccess
+
+
+@dataclass(frozen=True)
+class VisibilityScope:
+    """Область видимости справочников: `office_id is None` — вся система (ГО)."""
+
+    office_id: uuid.UUID | None
+
+    @property
+    def is_global(self) -> bool:
+        """Пользователь головного офиса — видит все записи (§10.1)."""
+        return self.office_id is None
+
+
+def scope_for(*, is_head_office: bool, regional_office_id: uuid.UUID) -> VisibilityScope:
+    """Область видимости по данным о филиале пользователя (§10.1): ГО — вся система, иначе — свой филиал.
+
+    ASSUMPTION(OQ-86): `business_segment` не учитывается.
+    """
+    return VisibilityScope(office_id=None if is_head_office else regional_office_id)
+
+
+def scope_of(principal: "UserWithAccess") -> VisibilityScope:
+    """Область видимости текущего пользователя (§10.1); роли и права `principal` не участвуют."""
+    return scope_for(
+        is_head_office=principal.regional_office.is_head_office,
+        regional_office_id=principal.user.regional_office_id,
+    )
+
+
+def company_groups_visible(stmt: Select[Any], scope: VisibilityScope) -> Select[Any]:
+    """Фильтр выборки ГК (§10.1): для филиала — `company_group.base_regional_office_id = <филиал>`.
+
+    `stmt` должен ссылаться на `CompanyGroup` (выборка ГК или запрос, уже присоединивший ГК). Для ГО `stmt` не меняется.
+    """
+    if scope.office_id is None:
+        return stmt
+    return stmt.where(CompanyGroup.base_regional_office_id == scope.office_id)
+
+
+def legal_entities_visible(stmt: Select[Any], scope: VisibilityScope) -> Select[Any]:
+    """Фильтр выборки ЮЛ (§10.1): для филиала — ЮЛ в ГК своего филиала ИЛИ монозаёмщик своего филиала.
+
+    Принадлежность ГК филиалу проверяется подзапросом `company_group_id IN (SELECT id … WHERE base_regional_office_id =
+    <филиал>)`, а не JOIN: число строк не меняется, а запрос вызывающего может сам присоединять `company_group` без
+    конфликта имён. Филиал самого ЮЛ в группе не важен — видимость определяет ГК. Для ГО `stmt` не меняется.
+    """
+    if scope.office_id is None:
+        return stmt
+    groups_of_office = select(CompanyGroup.id).where(CompanyGroup.base_regional_office_id == scope.office_id)
+    return stmt.where(
+        or_(
+            LegalEntity.company_group_id.in_(groups_of_office),
+            and_(LegalEntity.company_group_id.is_(None), LegalEntity.regional_office_id == scope.office_id),
+        )
+    )
+
+
+def company_group_is_visible(group: CompanyGroup, scope: VisibilityScope) -> bool:
+    """Видна ли ГК пользователю (§10.1); то же правило, что :func:`company_groups_visible`."""
+    return scope.office_id is None or group.base_regional_office_id == scope.office_id
+
+
+def legal_entity_is_visible(entity: LegalEntity, group_office_id: uuid.UUID | None, scope: VisibilityScope) -> bool:
+    """Виден ли ЮЛ пользователю (§10.1); то же правило, что :func:`legal_entities_visible`.
+
+    `group_office_id` — `base_regional_office_id` ГК этого ЮЛ (для монозаёмщика игнорируется).
+    """
+    if scope.office_id is None:
+        return True
+    if entity.company_group_id is not None:
+        return group_office_id == scope.office_id
+    return entity.regional_office_id == scope.office_id
+
+
+def office_for_create(scope: VisibilityScope, requested: uuid.UUID | None) -> uuid.UUID | None:
+    """Филиал закрепления новой ГК (spec directory-policy «Политика применяется до прав на изменение», §10.1).
+
+    ГО — `requested` как есть (в т.ч. `None`); филиал — `None` превращается в свой филиал, свой принимается, чужой →
+    422 `OFFICE_NOT_ALLOWED`.
+    """
+    if scope.office_id is None:
+        return requested
+    if requested is None or requested == scope.office_id:
+        return scope.office_id
+    raise DomainValidationError("OFFICE_NOT_ALLOWED", "ГК можно закрепить только за своим филиалом")

api/tests/test_domain_policy.py

@@ -0,0 +1,270 @@
+"""Policy видимости справочников (§10.1 `05-domain-rules.md`, spec directory-policy, task 2.1).
+
+Unit-часть — без БД: область видимости из данных о филиале, SQL-фильтры (компиляция `Select` под PostgreSQL с
+подставленными значениями), проверки отдельных объектов и выбор филиала при создании ГК.
+Integration-часть — семантика SQL-фильтров на настоящей БД: какие именно ГК и ЮЛ попадают в выборку, и что SQL-фильтр
+совпадает с проверкой отдельного объекта (`*_is_visible`).
+"""
+
+import re
+import uuid
+from types import SimpleNamespace
+from typing import Any, cast
+
+import pytest
+from fastapi import status as http_status
+from sqlalchemy import Select, func, select
+from sqlalchemy.dialects import postgresql
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.common.exceptions import DomainValidationError
+from src.domain.policy import (
+    VisibilityScope,
+    company_group_is_visible,
+    company_groups_visible,
+    legal_entities_visible,
+    legal_entity_is_visible,
+    office_for_create,
+    scope_for,
+    scope_of,
+)
+from src.models.directories import CompanyGroup, LegalEntity
+from tests.factories import make_company_group, make_legal_entity, make_regional_office
+
+OFFICE_X = uuid.UUID("00000000-0000-0000-0000-0000000000aa")
+OFFICE_Y = uuid.UUID("00000000-0000-0000-0000-0000000000bb")
+GLOBAL = VisibilityScope(office_id=None)
+BRANCH_X = VisibilityScope(office_id=OFFICE_X)
+
+
+def _sql(stmt: Select[Any]) -> str:
+    """SQL под PostgreSQL с подставленными значениями, пробелы схлопнуты в один."""
+    compiled = stmt.compile(dialect=postgresql.dialect(), compile_kwargs={"literal_binds": True})
+    return re.sub(r"\s+", " ", str(compiled)).strip()
+
+
+def _where(stmt: Select[Any]) -> str:
+    """Только часть после WHERE (или пустая строка)."""
+    sql = _sql(stmt)
+    return sql.split(" WHERE ", 1)[1] if " WHERE " in sql else ""
+
+
+# --- область видимости ---
+
+
+def test_head_office_scope_is_global() -> None:
+    scope = scope_for(is_head_office=True, regional_office_id=OFFICE_X)
+    assert scope.is_global
+    assert scope.office_id is None
+
+
+def test_branch_scope_is_bound_to_its_office() -> None:
+    scope = scope_for(is_head_office=False, regional_office_id=OFFICE_X)
+    assert not scope.is_global
+    assert scope.office_id == OFFICE_X
+
+
+@pytest.mark.parametrize(("is_head_office", "expected"), [(True, GLOBAL), (False, BRANCH_X)])
+def test_scope_of_principal_uses_office_data_not_role(is_head_office: bool, expected: VisibilityScope) -> None:
+    # Роль заведомо «чужая» уровню: область зависит только от филиала (права — данные, код роли не читается)
+    principal = SimpleNamespace(
+        user=SimpleNamespace(regional_office_id=OFFICE_X),
+        regional_office=SimpleNamespace(id=OFFICE_X, is_head_office=is_head_office),
+        roles=("KA_RF",) if is_head_office else ("KA_GO",),
+        permissions=frozenset({"DIRECTORY_VIEW"}),
+    )
+    assert scope_of(cast(Any, principal)) == expected
+
+
+# --- SQL-фильтр ГК ---
+
+
+def test_company_groups_global_returns_statement_unchanged() -> None:
+    stmt = select(CompanyGroup)
+    filtered = company_groups_visible(stmt, GLOBAL)
+    assert filtered is stmt
+    assert _where(filtered) == ""
+
+
+def test_company_groups_branch_filters_by_base_office() -> None:
+    filtered = company_groups_visible(select(CompanyGroup), BRANCH_X)
+    assert _where(filtered) == f"dkb.company_group.base_regional_office_id = '{OFFICE_X}'"
+
+
+def test_company_groups_filter_composes_with_existing_where() -> None:
+    stmt = select(CompanyGroup).where(CompanyGroup.is_archived.is_(False))
+    where = _where(company_groups_visible(stmt, BRANCH_X))
+    assert where == (
+        f"dkb.company_group.is_archived IS false AND dkb.company_group.base_regional_office_id = '{OFFICE_X}'"
+    )
+
+
+# --- SQL-фильтр ЮЛ ---
+
+
+def test_legal_entities_global_returns_statement_unchanged() -> None:
+    stmt = select(LegalEntity)
+    filtered = legal_entities_visible(stmt, GLOBAL)
+    assert filtered is stmt
+    assert _where(filtered) == ""
+
+
+def test_legal_entities_branch_filter_is_group_of_office_or_mono_borrower_of_office() -> None:
+    where = _where(legal_entities_visible(select(LegalEntity), BRANCH_X))
+    office = f"'{OFFICE_X}'"
+    # ЮЛ в ГК своего филиала ИЛИ (монозаёмщик И филиал ЮЛ — свой); единственное условие на запрос, обе ветви — по X
+    assert where == (
+        "dkb.legal_entity.company_group_id IN (SELECT dkb.company_group.id FROM dkb.company_group "
+        f"WHERE dkb.company_group.base_regional_office_id = {office}) "
+        f"OR dkb.legal_entity.company_group_id IS NULL AND dkb.legal_entity.regional_office_id = {office}"
+    )
+    assert " OR " in where
+    assert "IS NULL" in where
+
+
+def test_legal_entities_branch_filter_does_not_multiply_rows_or_clash_with_caller_join() -> None:
+    # Вызывающий сам присоединяет ГК (например, для имени группы): фильтр не добавляет второй FROM/JOIN company_group
+    stmt = select(LegalEntity, CompanyGroup.name).outerjoin(
+        CompanyGroup, CompanyGroup.id == LegalEntity.company_group_id
+    )
+    sql = _sql(legal_entities_visible(stmt, BRANCH_X))
+    from_clause = sql.split(" WHERE ", 1)[0].split(" FROM ", 1)[1]
+    assert from_clause == (
+        "dkb.legal_entity LEFT OUTER JOIN dkb.company_group ON dkb.company_group.id = dkb.legal_entity.company_group_id"
+    )
+
+
+def test_legal_entities_filter_works_for_count_statement() -> None:
+    stmt = select(func.count()).select_from(LegalEntity)
+    sql = _sql(legal_entities_visible(stmt, BRANCH_X))
+    assert sql.startswith("SELECT count(*) AS count_1 FROM dkb.legal_entity WHERE ")
+
+
+# --- проверка отдельной ГК ---
+
+
+@pytest.mark.parametrize(
+    ("scope", "group_office", "visible"),
+    [
+        (GLOBAL, OFFICE_Y, True),
+        (GLOBAL, None, True),
+        (BRANCH_X, OFFICE_X, True),
+        (BRANCH_X, OFFICE_Y, False),
+        (BRANCH_X, None, False),
+    ],
+)
+def test_company_group_is_visible(scope: VisibilityScope, group_office: uuid.UUID | None, visible: bool) -> None:
+    group = CompanyGroup(code="GK-000001", name="ГК", base_regional_office_id=group_office)
+    assert company_group_is_visible(group, scope) is visible
+
+
+# --- проверка отдельного ЮЛ ---
+
+
+@pytest.mark.parametrize(
+    ("scope", "in_group", "group_office", "entity_office", "visible"),
+    [
+        # монозаёмщик своего / чужого / без филиала
+        (BRANCH_X, False, None, OFFICE_X, True),
+        (BRANCH_X, False, None, OFFICE_Y, False),
+        (BRANCH_X, False, None, None, False),
+        # ЮЛ в ГК своего / чужого филиала; филиал самого ЮЛ не важен — решает ГК
+        (BRANCH_X, True, OFFICE_X, OFFICE_Y, True),
+        (BRANCH_X, True, OFFICE_Y, OFFICE_X, False),
+        (BRANCH_X, True, None, OFFICE_X, False),
+        # ГО видит всё
+        (GLOBAL, False, None, OFFICE_Y, True),
+        (GLOBAL, True, OFFICE_Y, OFFICE_Y, True),
+    ],
+)
+def test_legal_entity_is_visible(
+    scope: VisibilityScope,
+    in_group: bool,
+    group_office: uuid.UUID | None,
+    entity_office: uuid.UUID | None,
+    visible: bool,
+) -> None:
+    entity = LegalEntity(
+        inn="7700000000",
+        name="ООО Тест",
+        company_group_id=uuid.uuid4() if in_group else None,
+        regional_office_id=entity_office,
+    )
+    assert legal_entity_is_visible(entity, group_office, scope) is visible
+
+
+# --- филиал при создании ГК ---
+
+
+@pytest.mark.parametrize("requested", [None, OFFICE_Y])
+def test_office_for_create_head_office_keeps_requested(requested: uuid.UUID | None) -> None:
+    assert office_for_create(GLOBAL, requested) == requested
+
+
+def test_office_for_create_branch_defaults_to_own_office() -> None:
+    assert office_for_create(BRANCH_X, None) == OFFICE_X
+
+
+def test_office_for_create_branch_accepts_own_office() -> None:
+    assert office_for_create(BRANCH_X, OFFICE_X) == OFFICE_X
+
+
+def test_office_for_create_branch_rejects_foreign_office() -> None:
+    with pytest.raises(DomainValidationError) as exc_info:
+        office_for_create(BRANCH_X, OFFICE_Y)
+    assert exc_info.value.code == "OFFICE_NOT_ALLOWED"
+    assert exc_info.value.status == http_status.HTTP_422_UNPROCESSABLE_CONTENT
+    assert exc_info.value.message == "ГК можно закрепить только за своим филиалом"
+
+
+# --- семантика SQL-фильтров на настоящей БД ---
+
+
+@pytest.mark.integration
+async def test_sql_filters_match_object_checks_on_real_db(db_session: AsyncSession) -> None:
+    office_x = await make_regional_office(db_session)
+    office_y = await make_regional_office(db_session)
+    group_x = await make_company_group(db_session, name="ГК X", office=office_x)
+    group_y = await make_company_group(db_session, name="ГК Y", office=office_y)
+    group_none = await make_company_group(db_session, name="ГК без филиала")
+    entities = {
+        "в ГК X": await make_legal_entity(db_session, group=group_x),
+        "в ГК X, филиал ЮЛ Y": await make_legal_entity(db_session, group=group_x, office=office_y),
+        "в ГК Y": await make_legal_entity(db_session, group=group_y),
+        "в ГК Y, филиал ЮЛ X": await make_legal_entity(db_session, group=group_y, office=office_x),
+        "в ГК без филиала, филиал ЮЛ X": await make_legal_entity(db_session, group=group_none, office=office_x),
+        "монозаёмщик X": await make_legal_entity(db_session, office=office_x),
+        "монозаёмщик Y": await make_legal_entity(db_session, office=office_y),
+        "монозаёмщик без филиала": await make_legal_entity(db_session),
+    }
+    names_by_id = {entity.id: name for name, entity in entities.items()}
+    groups = {group.id: group for group in (group_x, group_y, group_none)}
+    branch_x = scope_for(is_head_office=False, regional_office_id=office_x.id)
+    head = scope_for(is_head_office=True, regional_office_id=office_x.id)
+
+    async def visible_entities(scope: VisibilityScope) -> set[str]:
+        stmt = legal_entities_visible(select(LegalEntity.id).where(LegalEntity.id.in_(names_by_id)), scope)
+        return {names_by_id[row] for row in (await db_session.scalars(stmt)).all()}
+
+    async def visible_groups(scope: VisibilityScope) -> set[uuid.UUID]:
+        stmt = company_groups_visible(select(CompanyGroup.id).where(CompanyGroup.id.in_(groups)), scope)
+        return set((await db_session.scalars(stmt)).all())
+
+    assert await visible_entities(branch_x) == {"в ГК X", "в ГК X, филиал ЮЛ Y", "монозаёмщик X"}
+    assert await visible_entities(head) == set(entities)
+    assert await visible_groups(branch_x) == {group_x.id}
+    assert await visible_groups(head) == set(groups)
+
+    # SQL-фильтр и проверка отдельного объекта согласованы
+    for scope in (branch_x, head):
+        expected_entities = {
+            name
+            for name, entity in entities.items()
+            if legal_entity_is_visible(
+                entity,
+                groups[entity.company_group_id].base_regional_office_id if entity.company_group_id else None,
+                scope,
+            )
+        }
+        assert await visible_entities(scope) == expected_entities
+        assert await visible_groups(scope) == {gid for gid, g in groups.items() if company_group_is_visible(g, scope)}

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

@@ -277,9 +277,9 @@ def office_for_create(scope: VisibilityScope, requested: uuid.UUID | None) -> uu
 ```
 `scope_of` импортирует `UserWithAccess` только под `TYPE_CHECKING`, чтобы `domain/` не тянул репозитории.
 
-- [ ] **Step 1: Тесты** (`tests/test_domain_policy.py`, без БД): `scope_for(is_head_office=True, …).is_global`; филиал → `office_id`; `company_groups_visible` для global возвращает тот же `stmt` (compile без WHERE), для филиала — `str(stmt)` содержит `base_regional_office_id`; `legal_entities_visible` содержит `OR` и `IS NULL`; `company_group_is_visible` — 4 комбинации; `legal_entity_is_visible` — монозаёмщик своего/чужого филиала, ЮЛ в ГК своего/чужого; `office_for_create` — 4 ветви, `DomainValidationError.code == "OFFICE_NOT_ALLOWED"`.
-- [ ] **Step 2:** падает → **Step 3:** реализация → **Step 4:** `pytest tests/test_domain_policy.py -v`, `mypy src` (strict для `src.domain.*`) → PASS.
-- [ ] **Step 5: Commit** `feat(domain): policy видимости справочников (§10.1, ASSUMPTION(OQ-86))`.
+- [x] **Step 1: Тесты** (`tests/test_domain_policy.py`, без БД): `scope_for(is_head_office=True, …).is_global`; филиал → `office_id`; `company_groups_visible` для global возвращает тот же `stmt` (compile без WHERE), для филиала — `str(stmt)` содержит `base_regional_office_id`; `legal_entities_visible` содержит `OR` и `IS NULL`; `company_group_is_visible` — 4 комбинации; `legal_entity_is_visible` — монозаёмщик своего/чужого филиала, ЮЛ в ГК своего/чужого; `office_for_create` — 4 ветви, `DomainValidationError.code == "OFFICE_NOT_ALLOWED"`.
+- [x] **Step 2:** падает → **Step 3:** реализация → **Step 4:** `pytest tests/test_domain_policy.py -v`, `mypy src` (strict для `src.domain.*`) → PASS.
+- [x] **Step 5: Commit** `feat(domain): policy видимости справочников (§10.1, ASSUMPTION(OQ-86))`.
 
 ---
 

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

@@ -26,7 +26,7 @@ test-integration`). Пути — от корня репозитория. Доп
 
 ## 2. Policy и справочники
 
-- [ ] 2.1 **Policy.** `api/src/domain/policy.py` (strict mypy): `VisibilityScope`, `scope_for(principal)`, `apply_scope_company_groups(stmt, scope)`,
+- [x] 2.1 **Policy.** `api/src/domain/policy.py` (strict mypy): `VisibilityScope`, `scope_for(principal)`, `apply_scope_company_groups(stmt, scope)`,
       `apply_scope_legal_entities(stmt, scope)`, `assert_office_allowed(scope, office_id)`; unit-тесты без БД на все ветки
       (ГО, филиал, монозаёмщик, чужой филиал → `OFFICE_NOT_ALLOWED`).
 - [ ] 2.2 **Филиалы и ГК.** `repositories/directories.py` (запросы с scope), `services/company_groups.py` (создание с `code` из