"""What did THIS MOVE change? — the question the packet never answered. Before this module the packet counted a mover's **inventory** and put that number on a card: 「63 项待你确认」. Nothing in the pipeline ever asked whether the move had anything to do with those 63 items. The one piece of code that computed a real diff — ``ProvisioningService.preview_role_change`` — resolved to *expected = ∅* because ``provision_policies`` is empty in production, which makes its answer "revoke everything" for everybody; and its return value was discarded by the caller anyway. So this module answers a narrower question, and refuses to answer it when it cannot: Of the things this person holds, which ones does the move actually call into question? Three signals, deliberately unequal. **policy** — a configured ``provision_policies`` rule for the role they moved INTO. The only source that unlocks an employee-facing push by default, because it is the only one that is a *statement of intent* rather than an observation. 🔴 The rule that makes this safe: a policy can only speak about a holding it could have been written about. ``policy_universe`` is every target any active policy mentions; a holding outside it gets NO policy verdict. Without that rule, a mover holding forty things in a department whose policies mention three seats would come back with thirty-seven "not expected" items — the inventory count again, wearing a delta's clothes. With it, an empty policy table yields an empty universe, no verdicts, and no push. That is the production state today and the gate holds by construction. **provenance** — the holding was granted while the person sat in the department they have just left. Not an opinion about what they *should* have; a fact about *why they got it*. Strong when available, and frequently unavailable: it needs a grant timestamp (only the access types carry one) and a department history (reconstructed from ``role_change`` lifecycle events, which only start when the coordinator shipped). Everything older than the history reads ``unknown``, never "fine". **peer** — how many of the mover's NEW peers hold the same thing. Display only, never a reason to interrupt somebody: a department where everybody over-holds looks exactly like a department where everybody is correctly provisioned. It ranks and explains; it does not accuse. Every item ends with a ``delta_kind``. Two of them count toward ``delta_items``; the rest exist so the page can say *why* something is not counted, which is the difference between "we checked and it is fine" and "we could not tell". """ from __future__ import annotations import logging from dataclasses import dataclass, field from datetime import UTC, date, datetime from typing import Any from uuid import UUID from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from app.domains.hr.models import ( HrLifecycleEvent, HrRoleChangeReview, HrRoleChangeReviewItem, ) from app.domains.provisioning.models import ProvisionPolicy from app.hr_lifecycle.mover_review_settings import ( BASELINE_NONE, BASELINE_PEER, BASELINE_POLICY, BASELINE_PROVENANCE, ) from app.identity.person_models import IdentityOrgMembership, IdentityPerson logger = logging.getLogger(__name__) ROLE_CHANGE_EVENT_TYPE = "role_change" # ── delta_kind vocabulary ─────────────────────────────────────────── # # Ordered by how much they claim. Only the two marked COUNTS feed # ``delta_items``; the rest are the vocabulary the page needs to # distinguish "checked, fine" from "could not tell". #: COUNTS. A policy for the new role exists and does not include this. DELTA_POLICY_UNEXPECTED = "policy_unexpected" #: COUNTS. Granted while the person sat in the department they left. DELTA_PROVENANCE_STALE = "provenance_stale" #: A policy for the new role explicitly includes this. Checked, fine. DELTA_POLICY_EXPECTED = "policy_expected" #: Granted after the move, or while already in the new department. DELTA_PROVENANCE_CURRENT = "provenance_current" #: Few of the new peers hold it. Ranking only — never counted. DELTA_PEER_RARE = "peer_rare" #: Most of the new peers hold it. Ranking only. DELTA_PEER_COMMON = "peer_common" #: No signal could speak about this holding. The honest default. DELTA_UNKNOWN = "unknown" #: The kinds that mean "this move calls this into question". COUNTED_DELTA_KINDS = (DELTA_POLICY_UNEXPECTED, DELTA_PROVENANCE_STALE) #: A holding is "rare among the new peers" below this share. 0.25 is a #: presentation threshold, not a decision threshold — nothing acts on #: it, so it only has to be useful for sorting. PEER_RARE_BELOW = 0.25 #: Below this many peers the ratio is noise (one colleague makes it #: 100%), so no peer verdict is issued at all. PEER_MIN_COHORT = 4 @dataclass(slots=True) class DeltaOutcome: """What the engine concluded about one packet.""" #: Holdings whose ``delta_kind`` is in ``COUNTED_DELTA_KINDS``. delta_items: int = 0 #: The strongest source that produced at least one counted item. #: ``none`` when nothing could speak — which is what keeps the push #: gate closed. baseline_source: str = BASELINE_NONE #: Per-kind tallies, for the page's "why is this number what it is". kinds: dict[str, int] = field(default_factory=dict) #: Targets the new role's policies expect that the person does NOT #: hold. Not items (items are things they hold) — a count, so IT can #: see a move created a provisioning gap as well as a review. policy_missing: int = 0 #: How much of the packet each signal was able to speak about, as a #: fraction of reviewable holdings. A delta of 0 against coverage of #: 0.02 is a different statement from a delta of 0 against 1.0. coverage: dict[str, float] = field(default_factory=dict) computed_at: datetime | None = None def to_json(self) -> dict[str, Any]: return { "delta_items": self.delta_items, "baseline_source": self.baseline_source, "kinds": dict(self.kinds), "policy_missing": self.policy_missing, "coverage": {k: round(v, 4) for k, v in self.coverage.items()}, "computed_at": ( self.computed_at.isoformat() if self.computed_at else None ), } # ── Department history ────────────────────────────────────────────── async def resolve_dept_history( db: AsyncSession, person_id: UUID, ) -> list[tuple[date, str | None]]: """``[(effective_date, dept_code_from_that_date)]``, oldest first. ``identity_org_memberships`` carries no validity interval — it is current state only — so the history has to be reconstructed from the ``role_change`` ledger: each event states the department on both sides of one transition. The list is deliberately open-ended at the FRONT: the first entry is the earliest event's ``before`` department, dated to that event, and anything granted before it is *unknown* rather than attributed to that department. The ledger starts when the coordinator shipped, and pretending it reaches further back would put a confident wrong department on a two-year-old grant. """ rows = ( await db.execute( select(HrLifecycleEvent) .where( HrLifecycleEvent.person_id == person_id, HrLifecycleEvent.event_type == ROLE_CHANGE_EVENT_TYPE, ) .order_by(HrLifecycleEvent.effective_date.asc()) ) ).scalars().all() history: list[tuple[date, str | None]] = [] for event in rows: detail = event.detail or {} before = (detail.get("before") or {}).get("dept_code") after = (detail.get("after") or {}).get("dept_code") if not history: history.append((event.effective_date, before)) history.append((event.effective_date, after)) return history def dept_at( history: list[tuple[date, str | None]], when: date, ) -> tuple[str | None, bool]: """``(dept_code, known)`` for a moment in this person's history. ``known`` is the whole point. ``(None, False)`` means "the ledger does not reach back this far", which must never be confused with ``(None, True)`` — "they demonstrably had no department". The provenance signal reports ``unknown`` for the first and can speak about the second. """ if not history: return None, False first_date, first_dept = history[0] if when < first_date: return None, False current = first_dept for eff, dept in history: if eff <= when: current = dept else: break return current, True # ── Policy signal ─────────────────────────────────────────────────── async def resolve_dept_name(db: AsyncSession, dept_code: str | None) -> str | None: """``org_node_source_id`` → ``org_node_name``. Load-bearing, and easy to get wrong: ``provision_policies`` .``department_name`` is compared against ``IdentityPerson`` .``primary_department``, which is a Python property returning the primary membership's **org_node_name** — a display name, not the source id the packet stores. Passing the code straight through matches no policy at all, and "no policy matched" is indistinguish- able from "no policy exists" unless you know to look. """ if not dept_code: return None name = await db.scalar( select(IdentityOrgMembership.org_node_name) .where( IdentityOrgMembership.org_node_source_id == dept_code, IdentityOrgMembership.deleted_at.is_(None), ) .limit(1) ) return str(name) if name else None async def resolve_dept_names( db: AsyncSession, codes: set[str], ) -> dict[str, str]: """``org_node_source_id`` → display name, for a whole set at once. Batched rather than per item: the provenance evidence names a department on every classified holding, and a packet is hundreds of them. """ if not codes: return {} rows = ( await db.execute( select( IdentityOrgMembership.org_node_source_id, IdentityOrgMembership.org_path, IdentityOrgMembership.org_node_name, ).where( IdentityOrgMembership.org_node_source_id.in_(codes), IdentityOrgMembership.deleted_at.is_(None), ).distinct() ) ).all() out: dict[str, str] = {} for code, path, name in rows: if code and code not in out: label = path or name if label: out[str(code)] = str(label)[:512] return out async def _primary_department_name( db: AsyncSession, person_id: UUID, ) -> str | None: """The person's current primary department NAME, by query. Deliberately not ``IdentityPerson.primary_department`` — see the call site: that property walks a relationship and raises MissingGreenlet on an expired instance. """ name = await db.scalar( select(IdentityOrgMembership.org_node_name) .where( IdentityOrgMembership.person_id == person_id, IdentityOrgMembership.is_primary.is_(True), IdentityOrgMembership.deleted_at.is_(None), ) .limit(1) ) return str(name) if name else None @dataclass(slots=True) class PolicySignal: """What ``provision_policies`` can say about this person's new role.""" #: Every target ANY active policy mentions. A holding outside this #: set gets no policy verdict — policies are silent about it, and #: silence is not a verdict. universe: set[str] = field(default_factory=set) #: Targets the policies matching this person's NEW attributes expect. expected: set[str] = field(default_factory=set) #: True when at least one policy matched this person. False means #: the policies exist but say nothing about this role, which is #: still "we cannot tell" rather than "they should have nothing". matched_any: bool = False async def compute_policy_signal( db: AsyncSession, *, department: str | None, job_title: str | None, location: str | None, ) -> PolicySignal: """Load the policy universe and the expectation for one role. Two queries, both bounded by the policy table (tens of rows at most), and deliberately NOT routed through ``ProvisioningService._match_policies_for_attrs``: that helper is used by the onboarding executor to decide what to GRANT, and it treats an empty match as "grant nothing" — a correct answer there and a catastrophic one here, where an empty match must mean "we do not know", not "revoke everything". The predicate is the same; the conclusion drawn from emptiness is not, and keeping them separate is what stops one being changed on behalf of the other. """ from app.core.provision_match import policy_match_conditions universe_rows = ( await db.execute( select(ProvisionPolicy.target_type, ProvisionPolicy.target_id) .where(ProvisionPolicy.is_active.is_(True)) ) ).all() universe = {f"{t}:{i}" for t, i in universe_rows} if not universe: # The production state on 2026-09-05. No policy exists, so no # policy can speak, so no packet can be pushed. Returned as an # empty signal rather than as an exception: an empty baseline is # an ordinary configuration, not a fault. return PolicySignal() conditions = policy_match_conditions( ProvisionPolicy, department=department, job_title=job_title, location=location, strict_null_on_missing=False, ) matched = ( await db.execute( select(ProvisionPolicy.target_type, ProvisionPolicy.target_id) .where(ProvisionPolicy.is_active.is_(True), *conditions) ) ).all() return PolicySignal( universe=universe, expected={f"{t}:{i}" for t, i in matched}, matched_any=bool(matched), ) # ── Peer signal ───────────────────────────────────────────────────── async def compute_peer_shares( db: AsyncSession, *, dept_code: str | None, exclude_person_id: UUID, ) -> tuple[dict[str, float], int]: """``({peer_key: share}, cohort_size)`` for the mover's NEW peers. Only the three holding kinds that have a stable cross-person identity (a plan, a permission, a cloud scope+role). Names are deliberately not used as keys: two plans can print the same string, and a rename would silently re-bucket everybody. Returns an empty map for a cohort below ``PEER_MIN_COHORT`` — with three colleagues every share is 33%, 67% or 100% and the number reads as evidence when it is noise. """ if not dept_code: return {}, 0 peer_ids = ( await db.execute( select(IdentityPerson.id) .join( IdentityOrgMembership, IdentityOrgMembership.person_id == IdentityPerson.id, ) .where( IdentityOrgMembership.org_node_source_id == dept_code, IdentityOrgMembership.is_primary.is_(True), IdentityOrgMembership.deleted_at.is_(None), IdentityPerson.deleted_at.is_(None), IdentityPerson.employment_status == "active", IdentityPerson.id != exclude_person_id, ) .distinct() ) ).scalars().all() cohort = len(peer_ids) if cohort < PEER_MIN_COHORT: return {}, cohort from app.domains.asset.models import SoftwareSeatAllocation from app.domains.cmdb.models import CmdbCloudAccessGrant from app.domains.permission.models import PERM_ACTIVE, UserPermission counts: dict[str, int] = {} for plan_id, n in ( await db.execute( select( SoftwareSeatAllocation.plan_id, func.count(func.distinct(SoftwareSeatAllocation.user_id)), ) .where( SoftwareSeatAllocation.user_id.in_(peer_ids), SoftwareSeatAllocation.status == "active", ) .group_by(SoftwareSeatAllocation.plan_id) ) ).all(): counts[f"seat:{plan_id}"] = int(n) for perm_id, n in ( await db.execute( select( UserPermission.permission_id, func.count(func.distinct(UserPermission.user_id)), ) .where( UserPermission.user_id.in_(peer_ids), UserPermission.status == PERM_ACTIVE, ) .group_by(UserPermission.permission_id) ) ).all(): counts[f"permission:{perm_id}"] = int(n) # ``person_id`` on cloud grants is a CHAR(36) string column with no # GUID adapter — bind strings or the IN matches nothing, silently # (memory: reference_asyncmy_uuid_bind). peer_strs = [str(p) for p in peer_ids] for scope_id, role_key, n in ( await db.execute( select( CmdbCloudAccessGrant.scope_id, CmdbCloudAccessGrant.role_key, func.count(func.distinct(CmdbCloudAccessGrant.person_id)), ) .where( CmdbCloudAccessGrant.person_id.in_(peer_strs), CmdbCloudAccessGrant.status.in_(("active", "pending")), ) .group_by( CmdbCloudAccessGrant.scope_id, CmdbCloudAccessGrant.role_key, ) ) ).all(): counts[f"cloud:{scope_id}:{role_key}"] = int(n) return {k: v / cohort for k, v in counts.items()}, cohort # ── The engine ────────────────────────────────────────────────────── async def annotate_review( db: AsyncSession, review: HrRoleChangeReview, ) -> DeltaOutcome: """Classify every reviewable item on one packet and tally the result. Writes ``delta_kind`` / ``delta_evidence`` on each item and ``delta_items`` / ``baseline_source`` on the packet. Idempotent: a re-run reclassifies from the current state of the world, which is the point — a packet built while the policy table was empty becomes meaningful the day policies are configured, without anybody re-triggering anything. Never raises for a data reason. A packet that cannot be classified is a packet whose delta is 0 and whose baseline is ``none``, which is exactly the state that keeps it un-pushable. """ from app.hr_lifecycle.mover_holdings import REVIEW as _REVIEW person = await db.get(IdentityPerson, review.person_id) items = ( await db.execute( select(HrRoleChangeReviewItem).where( HrRoleChangeReviewItem.review_id == review.id, ) ) ).scalars().all() reviewable = [i for i in items if i.disposition == _REVIEW] policy = await compute_policy_signal( db, # NAME, not the source id the packet stores — see # ``resolve_dept_name``. Falls back to the person's current # primary department, which at packet-build time IS the new one. # # 🔴 Resolved by QUERY, never through # ``IdentityPerson.primary_department``. That property walks the # ``org_memberships`` relationship, and although the mapping # declares ``lazy="selectin"``, an instance fetched with # ``db.get`` after a commit is expired: touching the property # then attempts IO from sync code and raises MissingGreenlet. # It only fires on THIS branch — when the packet's dept code has # no membership row — so it survived every earlier test and # crashed the nightly re-annotation sweep instead. department=( await resolve_dept_name(db, review.after_dept_code) or await _primary_department_name(db, review.person_id) ), job_title=review.after_job_title or (person.job_title if person else None), location=person.work_location_code if person else None, ) history = await resolve_dept_history(db, review.person_id) # One lookup for every department the history mentions, so the # per-item evidence can name them instead of printing codes. dept_names = await resolve_dept_names( db, {code for _at, code in history if code}, ) peer_shares, cohort = await compute_peer_shares( db, dept_code=review.after_dept_code, exclude_person_id=review.person_id, ) left_dept = review.before_dept_code moved_dept = bool(left_dept) and left_dept != review.after_dept_code kinds: dict[str, int] = {} spoke = {BASELINE_POLICY: 0, BASELINE_PROVENANCE: 0, BASELINE_PEER: 0} counted = 0 sources_that_counted: set[str] = set() for item in reviewable: detail = item.detail or {} evidence: dict[str, Any] = {} kind = DELTA_UNKNOWN # ── policy, strongest ────────────────────────────────────── # # 🔴 TWO conditions, and the second is easy to lose. # # ``universe`` says policies could have been written about this # KIND of holding. ``matched_any`` says policies were written # about THIS ROLE. Both are required, because "no policy matched # this person" means *we do not know what this role should # have* — not *this role should have nothing*. # # Without the second condition, somebody moving into a # department that has no policies at all gets ``expected = ∅`` # and every holding inside the global universe comes back # ``policy_unexpected``. That is the "revoke everything" shape # from ``preview_role_change``, surviving at role granularity # instead of table granularity — the same bug one level down. target = detail.get("policy_target") if ( target and policy.matched_any and policy.universe and target in policy.universe ): spoke[BASELINE_POLICY] += 1 if target in policy.expected: kind = DELTA_POLICY_EXPECTED evidence["policy"] = "expected_for_new_role" else: kind = DELTA_POLICY_UNEXPECTED evidence["policy"] = "not_expected_for_new_role" # ── provenance, only if policy was silent ────────────────── if kind == DELTA_UNKNOWN and moved_dept: raw = detail.get("granted_at") granted_on = _as_date(raw) if granted_on is not None: at, known = dept_at(history, granted_on) if known: spoke[BASELINE_PROVENANCE] += 1 evidence["granted_at"] = raw # A department NAME, not the source id. The reason # line reads 「授予时这个人还在「{dept}」」 straight to # a manager, and `2730378` in that sentence is the # same unreadability the console's JSON columns had # — fixed there, and it would have crept back in # here. The code is kept alongside for forensics. evidence["dept_when_granted_code"] = at evidence["dept_when_granted"] = ( dept_names.get(at) or at if at else None ) if at and at == left_dept: kind = DELTA_PROVENANCE_STALE else: kind = DELTA_PROVENANCE_CURRENT else: # The ledger does not reach this far back. Recorded # so the page can say WHY it is unknown rather than # leaving a bare "unknown". evidence["provenance"] = "before_recorded_history" # ── peer, ranking only, never overrides a verdict ────────── peer_key = detail.get("peer_key") if peer_key and peer_key in peer_shares: share = peer_shares[peer_key] spoke[BASELINE_PEER] += 1 evidence["peer_share"] = round(share, 4) evidence["peer_cohort"] = cohort if kind == DELTA_UNKNOWN: kind = ( DELTA_PEER_RARE if share < PEER_RARE_BELOW else DELTA_PEER_COMMON ) item.delta_kind = kind item.delta_evidence = evidence or None kinds[kind] = kinds.get(kind, 0) + 1 if kind in COUNTED_DELTA_KINDS: counted += 1 sources_that_counted.add( BASELINE_POLICY if kind == DELTA_POLICY_UNEXPECTED else BASELINE_PROVENANCE ) # Policy gaps: expected by the new role, not held. Not items — items # are things the person HAS — but IT should see that a move created a # provisioning gap as well as a review. held_targets = { (i.detail or {}).get("policy_target") for i in reviewable if (i.detail or {}).get("policy_target") } policy_missing = len(policy.expected - held_targets) total = len(reviewable) or 1 outcome = DeltaOutcome( delta_items=counted, # Strongest source that actually produced a counted item. Peer is # absent from this by construction — it can never be the reason a # packet becomes pushable. baseline_source=( BASELINE_POLICY if BASELINE_POLICY in sources_that_counted else BASELINE_PROVENANCE if BASELINE_PROVENANCE in sources_that_counted else BASELINE_NONE ), kinds=kinds, policy_missing=policy_missing, coverage={k: v / total for k, v in spoke.items()}, computed_at=datetime.now(UTC), ) review.delta_items = outcome.delta_items review.baseline_source = outcome.baseline_source review.delta_summary = outcome.to_json() review.delta_computed_at = outcome.computed_at return outcome def _as_date(raw: Any) -> date | None: if not raw: return None if isinstance(raw, date) and not isinstance(raw, datetime): return raw if isinstance(raw, datetime): return raw.date() try: return datetime.fromisoformat(str(raw).replace("Z", "+00:00")).date() except (TypeError, ValueError): return None __all__ = [ "COUNTED_DELTA_KINDS", "DELTA_PEER_COMMON", "DELTA_PEER_RARE", "DELTA_POLICY_EXPECTED", "DELTA_POLICY_UNEXPECTED", "DELTA_PROVENANCE_CURRENT", "DELTA_PROVENANCE_STALE", "DELTA_UNKNOWN", "DeltaOutcome", "PolicySignal", "annotate_review", "compute_peer_shares", "compute_policy_signal", "dept_at", "resolve_dept_history", "resolve_dept_name", "resolve_dept_names", ]