"""Person lifecycle-state scope — the single source of truth for "who counts as a current report / an employed person / a separated person". Why this module exists ----------------------- Helix has no first-class notion of "person eligibility state". The person lifecycle is three orthogonal axes — * ``employment_status`` ∈ {``active``, ``leaver_pending``, ``left``, ``pre_hire``} (HR employment state), * ``deleted_at`` (soft-delete / row visibility), * ``hris_record_state`` (present vs purged from the upstream HRIS), — and for a long time **every consumer re-derived "is this person one of my reports?" inline**, each with its own ``employment_status`` predicate. They drifted. The most visible symptom was the recurring "a departed report still shows in 我的团队 / 团队资源 / 全局搜索 / 关系图谱" bug class (PR #782 → #784 → #786 → team-footprint → GlobalSearch → RelationshipGraph — the *same* root cause patched seven times). A silent inverse also existed: the mobile aggregate team helper applied **no** ``employment_status`` filter at all, so mobile managers saw ``left`` reports indefinitely. The fix is structural, not another point-patch: one canonical BFS (:func:`subordinate_person_ids`) parameterised by an explicit :class:`SubordinateScope`, plus a composable :func:`employment_status_clause` for the non-subordinate query sites. A CI guard (``scripts/check_subordinate_helpers.py``) forbids new hand-rolled ``IdentityPerson.manager_person_id.in_(...)`` BFS loops so the bug class cannot reappear in the next surface someone adds. Scope semantics --------------- Scope is defined by which ``employment_status`` values it **excludes** from the *returned* set. Traversal always walks *through* every non-deleted node regardless of scope, so a ``left`` / ``leaver_pending`` intermediate manager never hides their still-active deeper reports (A → B(leaving) → C: A must still reach C). Only membership of the returned set is governed by the scope: * :attr:`SubordinateScope.MANAGED` — excludes ``left`` only. The "current team" view: people you still manage, including anyone in the offboarding window (``leaver_pending``) who is exactly who "即将离职" surfaces. Equivalent to the old ``!= 'left'`` filter. * :attr:`SubordinateScope.ACTIVE_ONLY` — excludes ``left`` **and** ``leaver_pending``. Stable holders for resource / cost attribution, where someone on their way out should not be counted. Equivalent to the old ``NOT IN ('left', 'leaver_pending')`` filter. * :attr:`SubordinateScope.ALL` — no ``employment_status`` filter (``deleted_at IS NULL`` only). Forensic / audit surfaces and the cloud-resource visibility views where a manager intentionally needs to see what a departed report still holds (those views carry their own ``include_left`` toggle). Equivalent to the legacy :func:`app.core.rbac.get_subordinate_person_ids` behaviour. ``IdentityPerson.id`` is a native SQLAlchemy ``Uuid`` column, so :func:`subordinate_person_ids` returns ``list[uuid.UUID]`` — the exact type the column yields and the type every ORM-row consumer (asset / dashboard / mobile / me_universe) compares against (their ``assigned_to`` query params and ``current_user.id`` are ``UUID``). The legacy :func:`app.core.rbac.get_subordinate_person_ids` wrapper keeps its historical ``list[str]`` contract for the cmdb / mcp / permission callers by stringifying the result. """ from __future__ import annotations import enum import uuid from typing import TYPE_CHECKING from sqlalchemy import select, true from sqlalchemy.sql.elements import ColumnElement from app.identity.person_models import IdentityPerson if TYPE_CHECKING: from sqlalchemy.ext.asyncio import AsyncSession # ── Canonical employment_status values ─────────────────────────────── # Mirror of ``IdentityPerson._ALLOWED_EMPLOYMENT_STATUSES`` (kept public # here so query sites compose against named constants, not string # literals). The ORM ``@validates`` on the model is the enforcement # point; these are for read-side predicates. EMPLOYMENT_ACTIVE = "active" EMPLOYMENT_LEAVER_PENDING = "leaver_pending" EMPLOYMENT_LEFT = "left" EMPLOYMENT_PRE_HIRE = "pre_hire" #: Still on payroll (employed today), whether or not offboarding has begun. EMPLOYED_STATUSES: frozenset[str] = frozenset( {EMPLOYMENT_ACTIVE, EMPLOYMENT_LEAVER_PENDING} ) #: Has departed — the row is retained (manager pointer, audit trail) but #: the person is no longer part of any current team / headcount. SEPARATED_STATUSES: frozenset[str] = frozenset({EMPLOYMENT_LEFT}) # Bound on BFS depth. The org is ~500 people; a real reporting chain is # 6–7 deep at most, so 10 covers every legitimate tree while still # capping a pathological / cycle-corrupted ``manager_person_id`` graph. # Cycle-safety is additionally guaranteed by the ``visited`` set. DEFAULT_MAX_DEPTH = 10 class SubordinateScope(enum.StrEnum): """Which employment states are *included* in a resolved report set. See the module docstring for the full contract. ``StrEnum`` so the value is log-/JSON-friendly. """ MANAGED = "managed" # exclude {left} ACTIVE_ONLY = "active_only" # exclude {left, leaver_pending} ALL = "all" # exclude {} #: Statuses excluded from the *returned* set per scope. Traversal is #: unaffected (always walks through every non-deleted node). _EXCLUDED_STATUSES: dict[SubordinateScope, frozenset[str]] = { SubordinateScope.MANAGED: SEPARATED_STATUSES, SubordinateScope.ACTIVE_ONLY: frozenset( {EMPLOYMENT_LEFT, EMPLOYMENT_LEAVER_PENDING} ), SubordinateScope.ALL: frozenset(), } def employment_status_clause( scope: SubordinateScope = SubordinateScope.MANAGED, *, col: ColumnElement[str | None] = IdentityPerson.employment_status, ) -> ColumnElement[bool]: """Composable ``WHERE`` predicate excluding the scope's barred states. Use at any query site that filters people by lifecycle state instead of hand-writing ``employment_status != 'left'`` — e.g.:: select(IdentityPerson).where( employment_status_clause(SubordinateScope.MANAGED), IdentityPerson.deleted_at.is_(None), ) ``SubordinateScope.ALL`` returns a no-op ``TRUE`` so the clause can be dropped unconditionally into a filter list. ``col`` lets the predicate target a different table's status column (e.g. an aliased / joined person) while keeping the exclusion set centralised. NOTE on NULL: ``col.notin_((...))`` evaluates to NULL → row excluded when ``employment_status`` is NULL, matching the long-standing ``!= 'left'`` semantics every call site already relied on. The ORM validator makes NULL unreachable for freshly written rows. """ excluded = _EXCLUDED_STATUSES[scope] if not excluded: return true() return col.notin_(tuple(sorted(excluded))) async def subordinate_person_ids( session: AsyncSession, manager_person_id: uuid.UUID | str, *, scope: SubordinateScope = SubordinateScope.MANAGED, max_depth: int = DEFAULT_MAX_DEPTH, ) -> list[uuid.UUID]: """Canonical subordinate resolver — the only manager-tree BFS. Returns the ``person_id`` (as ``uuid.UUID``) of every direct and transitive report of ``manager_person_id`` whose ``employment_status`` is permitted by ``scope``. The manager themselves is **excluded** (every caller adds ``self`` to its own IN-clause). Traversal walks through every ``deleted_at IS NULL`` node regardless of scope, so a ``left`` / ``leaver_pending`` intermediate manager does not truncate the subtree below them; only set membership is filtered. Bounded by ``max_depth`` and de-duplicated via ``visited`` so a corrupted (cyclic) ``manager_person_id`` graph cannot loop forever. Pass ``max_depth=1`` for direct reports only (the "my team" home surfaces). ``manager_person_id`` is accepted as ``UUID`` or ``str`` and normalised to ``UUID`` so dedup and the returned set are type-consistent with the native ``Uuid`` primary key. """ root = ( manager_person_id if isinstance(manager_person_id, uuid.UUID) else uuid.UUID(str(manager_person_id)) ) visited: set[uuid.UUID] = {root} frontier: list[uuid.UUID] = [root] excluded = _EXCLUDED_STATUSES[scope] out: list[uuid.UUID] = [] for _ in range(max_depth): if not frontier: break rows = await session.execute( select( IdentityPerson.id, IdentityPerson.employment_status, ).where( IdentityPerson.manager_person_id.in_(frontier), IdentityPerson.deleted_at.is_(None), ) ) next_frontier: list[uuid.UUID] = [] for pid, status in rows.all(): if pid in visited: continue visited.add(pid) # Always traverse through (subtree below a leaving manager is # still part of the org); only gate set membership on scope. next_frontier.append(pid) if status not in excluded: out.append(pid) frontier = next_frontier return out def includes_status(scope: SubordinateScope, employment_status: str | None) -> bool: """True when a person with ``employment_status`` belongs to ``scope``'s returned set. The in-Python mirror of :func:`employment_status_clause`, for callers that already hold the person object. Mirrors the SQL clause's NULL handling exactly: ``employment_status IS NULL`` is EXCLUDED for any scope that bars statuses (the SQL ``col.notin_((...))`` evaluates to NULL → row dropped), and INCLUDED only for ``ALL`` (whose clause is a no-op ``TRUE``). """ excluded = _EXCLUDED_STATUSES[scope] if not excluded: return True return employment_status is not None and employment_status not in excluded __all__ = [ "DEFAULT_MAX_DEPTH", "EMPLOYED_STATUSES", "EMPLOYMENT_ACTIVE", "EMPLOYMENT_LEAVER_PENDING", "EMPLOYMENT_LEFT", "EMPLOYMENT_PRE_HIRE", "SEPARATED_STATUSES", "SubordinateScope", "employment_status_clause", "includes_status", "subordinate_person_ids", ]