"""Configuration for the mover (role-change) review feature. Why this module exists at all: until 2026-09-05 the whole feature was governed by two ``scheduled_jobs.enabled`` flags and nothing else, and the thing those flags could not express is the one thing that actually matters — **whether Helix is in a position to say anything useful about a move.** The production finding that forced this: ``provision_policies`` holds **zero rows**. That table is the only baseline the platform has for "what should somebody in this department have", so the entitlement diff (``ProvisioningService.preview_role_change``) resolves to *expected = ∅* for every employee, which makes its answer "revoke everything this person holds" — for everybody, always. The review packet never used that diff; it listed the mover's entire **inventory** instead. So a manager's card said "238 items need your call" when the honest number was unknown and probably zero. Nine such cards went to the CTO and CEO on one morning; 223 items were listed and **zero** decisions were ever recorded. So the rule this module encodes, and which every push path must ask before it sends anything: 没有基线,就不要推送。 ``employee_push_enabled`` is the master switch and it defaults to **False**. ``require_baseline_for_push`` (default True) is the invariant behind it: even with the switch on, a packet may only be pushed when its delta was computed against a real baseline. The two together mean an operator cannot re-enable notifications by flipping a cron row — the send path re-reads this setting and refuses. Everything here is a *policy* knob, not a feature flag hiding half-built code: the review packet, the pages, the decisions and the audit trail all work with pushes off. What is off is only the unsolicited interruption. """ from __future__ import annotations from dataclasses import dataclass, field from typing import Any from sqlalchemy.ext.asyncio import AsyncSession from app.core.exceptions import AppException from app.integrations.models import PlatformSetting from app.integrations.service import get_platform_setting SETTING_KEY = "role_change.review" SETTING_DESCRIPTION = ( "转岗复核(mover review)的推送与差异口径。" "employee_push_enabled 是员工侧推送总开关,默认关 —— 2026-09-05 拍板:" "「我没有弄基线,你就不要推送」。require_baseline_for_push 是它背后的不变量:" "即使开关打开,一份没有基线来源的包裹也不允许推送(baseline_source='none' 时" "delta 无意义,卡片只能报存量,而存量不是这次转岗造成的)。" "allowed_baseline_sources 决定哪些基线来源算数 —— peer(同侪分布)是统计信号" "不是政策,默认不解锁推送。" ) #: Baseline sources, strongest first. #: #: ``policy`` — ``provision_policies`` matched on the person's NEW #: department / title / location. A configured, human-authored #: statement of what this role should have. The only source that is #: allowed to unlock a push by default. #: #: ``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, but it can only speak #: about holdings whose grant time can be located in the org history. #: #: ``peer`` — how common this holding is among their new peers. Useful #: for ranking, never sufficient on its own: it is a description of the #: status quo, and a department where everybody over-holds looks exactly #: like a department where everybody is correctly provisioned. BASELINE_POLICY = "policy" BASELINE_PROVENANCE = "provenance" BASELINE_PEER = "peer" BASELINE_NONE = "none" BASELINE_SOURCES: tuple[str, ...] = ( BASELINE_POLICY, BASELINE_PROVENANCE, BASELINE_PEER, ) #: Sources that may unlock an employee-facing push when #: ``employee_push_enabled`` is on. ``peer`` is deliberately absent. DEFAULT_PUSH_BASELINE_SOURCES: tuple[str, ...] = ( BASELINE_POLICY, BASELINE_PROVENANCE, ) @dataclass(slots=True) class RoleChangeReviewSettings: """Resolved ``role_change.review`` policy.""" #: Master switch for anything that reaches an ordinary employee #: (the new manager). Default OFF. employee_push_enabled: bool = False #: Even with the switch on, refuse to push a packet whose delta was #: not computed against a baseline. Turning this off is how somebody #: would re-create the "238 items, all of them inventory" card, so #: it is validated as a pair with the switch below. require_baseline_for_push: bool = True #: Which ``baseline_source`` values count as "a baseline" for the #: rule above. push_baseline_sources: list[str] = field( default_factory=lambda: list(DEFAULT_PUSH_BASELINE_SOURCES), ) #: A packet with zero delta items is not worth anyone's attention, #: so it is never pushed. Kept configurable because an org that has #: just configured its first policies may want to see the first #: batch regardless. min_delta_items_to_push: int = 1 #: The 30-day "still pending_review" chase for cloud grants. Same #: default-off treatment: it fans out to the scope owner AND every #: active global admin. stale_grant_push_enabled: bool = False stale_grant_days: int = 30 @classmethod def from_setting(cls, raw: Any) -> RoleChangeReviewSettings: """Build from the stored JSON, tolerating absence and junk. A malformed value must not take the send path down — but it must also never fail OPEN. Every unreadable field falls back to the dataclass default, and every default is the quiet one. """ if not isinstance(raw, dict): return cls() data = dict(raw) def _bool(key: str, fallback: bool) -> bool: value = data.get(key, fallback) return value if isinstance(value, bool) else fallback def _int(key: str, fallback: int) -> int: value = data.get(key, fallback) # ``bool`` is an ``int`` subclass — excluded first, or a # stray ``true`` in an int field would silently become 1. if isinstance(value, bool) or not isinstance(value, int): return fallback return int(value) sources = data.get("push_baseline_sources") if isinstance(sources, list): cleaned = [s for s in sources if s in BASELINE_SOURCES] # An empty / fully-invalid list means "nothing unlocks a # push". That is a valid, and safe, configuration — do not # silently restore the defaults over it. push_sources = cleaned if sources == [] or cleaned else list( DEFAULT_PUSH_BASELINE_SOURCES, ) else: push_sources = list(DEFAULT_PUSH_BASELINE_SOURCES) return cls( employee_push_enabled=_bool("employee_push_enabled", False), require_baseline_for_push=_bool("require_baseline_for_push", True), push_baseline_sources=push_sources, min_delta_items_to_push=_int("min_delta_items_to_push", 1), stale_grant_push_enabled=_bool("stale_grant_push_enabled", False), stale_grant_days=_int("stale_grant_days", 30), ) def to_setting(self) -> dict[str, Any]: return { "employee_push_enabled": self.employee_push_enabled, "require_baseline_for_push": self.require_baseline_for_push, "push_baseline_sources": list(self.push_baseline_sources), "min_delta_items_to_push": self.min_delta_items_to_push, "stale_grant_push_enabled": self.stale_grant_push_enabled, "stale_grant_days": self.stale_grant_days, } def validate(self) -> None: problems: list[str] = [] if self.min_delta_items_to_push < 1: problems.append("min_delta_items_to_push must be at least 1") if not 7 <= self.stale_grant_days <= 365: problems.append("stale_grant_days must be between 7 and 365") for source in self.push_baseline_sources: if source not in BASELINE_SOURCES: problems.append(f"unknown baseline source {source!r}") if self.employee_push_enabled and not self.require_baseline_for_push: # The exact configuration that produced the September 5th # cards. Refused rather than warned about: the reason it # went out was that nothing refused it. problems.append( "employee_push_enabled with require_baseline_for_push=false " "would push inventory counts as if they were change counts — " "the 2026-09-05 regression. Configure a baseline instead.", ) if ( self.employee_push_enabled and self.require_baseline_for_push and not self.push_baseline_sources ): problems.append( "employee_push_enabled with no push_baseline_sources can never " "send anything — either add a source or turn the switch off", ) if problems: raise AppException( code="ROLE_CHANGE_REVIEW_SETTINGS_INVALID", message="; ".join(problems), status_code=400, ) def may_push(self, *, baseline_source: str | None, delta_items: int) -> bool: """The single question every employee-facing send path asks. Kept as one method rather than re-derived at each call site: the rule that has to hold is "no baseline ⇒ no push", and a rule spelled out in three places is a rule with three chances to drift. """ if not self.employee_push_enabled: return False if delta_items < self.min_delta_items_to_push: return False if not self.require_baseline_for_push: return True return (baseline_source or BASELINE_NONE) in self.push_baseline_sources async def load_review_settings(db: AsyncSession) -> RoleChangeReviewSettings: raw = await get_platform_setting(SETTING_KEY, default=None, session=db) return RoleChangeReviewSettings.from_setting(raw) async def save_review_settings( db: AsyncSession, settings: RoleChangeReviewSettings, ) -> RoleChangeReviewSettings: settings.validate() row = await db.get(PlatformSetting, SETTING_KEY) if row is None: row = PlatformSetting( key=SETTING_KEY, value=settings.to_setting(), description=SETTING_DESCRIPTION, ) db.add(row) else: row.value = settings.to_setting() await db.flush() return settings __all__ = [ "BASELINE_NONE", "BASELINE_PEER", "BASELINE_POLICY", "BASELINE_PROVENANCE", "BASELINE_SOURCES", "DEFAULT_PUSH_BASELINE_SOURCES", "SETTING_DESCRIPTION", "SETTING_KEY", "RoleChangeReviewSettings", "load_review_settings", "save_review_settings", ]