"""转岗复核:停掉两条推送 + 给已积压的包裹盖「withheld」戳 + 落 role_change.review 配置。 生产实测(2026-09-05,TiDB,542 在职): * ``provision_policies`` = **0 行**。这是平台唯一的「某个部门/岗位应该有什么」基线, 所以 ``ProvisioningService.preview_role_change`` 算出来的 expected 恒为空集, 它的结论对每个人都是「把你现有的权限全部回收」。而且它的返回值在 ``role_change_coordinator`` 里根本没人接 —— 算完就丢。 * 于是复核包裹报的是这个人的**存量**(24 类持有物),不是这次转岗造成的**增量**。 * 9/5 09:30 真的发出去 9 张卡:CTO 5 张(63/15/13/13/11 项)、Jie Zhan 3 张、 CEO 1 张。全部 delivered(飞书 + 邮件)。收件人里 3 位只有 ``employee`` 角色, 卡片 CTA 指向 ``/admin/role-changes``(admin-only 路由),点了会被弹回首页。 * 223 个 item,``decision`` 全是 NULL —— 上线以来 0 次决断。 这条迁移做四件事,都为了「不打扰员工」: 1. **停两条推送 cron**(``role_change.review_notify`` / ``role_change.scope_review_sweep``)。 固化成迁移而不是只手工改库:否则新环境 seed 出来又是开的。 2. **给积压盖戳**。``hr_role_change_reviews`` 新增 ``notify_suppressed_at`` / ``notify_suppressed_reason``,并把现存所有 ``notified_at IS NULL`` 的包裹标成 withheld。不复用 ``notified_at``:那个字段的语义是「真的发过」,污染它以后就 分不清「通知过」和「被拦下」。没有这一步,任何一次重新启用都会在当天把整个 积压一次性轰出去。 3. **加 ``baseline_source`` / ``delta_items`` 两列**(默认 ``'none'`` / ``0``)。 发送路径问的是 ``settings.may_push(baseline_source, delta_items)``,所以在差异 引擎落地之前,**任何一份包裹都推不出去** —— 不管管理员把几个开关打开。把这个 不变量放到后面的 PR 才建立,中间就存在一个「翻一个 bool 就能重演 9/5」的窗口。 4. **落 ``platform_settings['role_change.review']``**,把「没有基线就不要推送」 写成可读可审的配置而不是一句口头约定。``employee_push_enabled`` 默认 False, ``require_baseline_for_push`` 默认 True,``push_baseline_sources`` 不含 peer (同侪分布是统计信号,不是政策)。发送路径每次都重读它 —— 光把 cron 打开 发不出任何东西。 向下兼容:suppression 两列可空;delta 两列 NOT NULL 但带 server_default,所以既有行 被填成 ``('none', 0)``(= 推不出去,正是想要的初值),旧代码不写这两列也照样 INSERT 成功。配置行按 key 存在性守卫;cron 只 UPDATE 不 INSERT(行不存在就什么都不做)。 重复执行是空转。downgrade **不**把 cron 重新打开 —— schema 回滚不等于基线问题解决了。 """ from __future__ import annotations import json import sqlalchemy as sa from alembic import op revision = "610" # 609 (leaver safety-net registry) landed on origin/main while this # branch was being committed — the note that used to sit here predicted # exactly that and said to re-point. Done. Re-check once more before # merging: four sessions are shipping migrations the same day, and the # repo rule is to claim the number against origin/main rather than # against your own worktree (two branches that both took 462 landed # hours apart and left the chain with two heads). down_revision = "609" branch_labels = None depends_on = None REVIEWS = "hr_role_change_reviews" SETTINGS = "platform_settings" JOBS = "scheduled_jobs" NOTIFY_JOB = "role_change.review_notify" SWEEP_JOB = "role_change.scope_review_sweep" SETTING_KEY = "role_change.review" #: Stamped on the backlog. Read by nobody programmatically — it exists #: so a human reading the row in six months learns why the packet was #: never sent without having to find this migration. SUPPRESS_REASON = "baseline_absent_2026_09_05" _SETTING_VALUE = { "employee_push_enabled": False, "require_baseline_for_push": True, "push_baseline_sources": ["policy", "provenance"], "min_delta_items_to_push": 1, "stale_grant_push_enabled": False, "stale_grant_days": 30, "admin_digest_enabled": True, } _SETTING_DESC = ( "转岗复核(mover review)的推送与差异口径。" "employee_push_enabled 是员工侧推送总开关,默认关 —— 2026-09-05 拍板:" "「我没有弄基线,你就不要推送」。require_baseline_for_push 是它背后的不变量:" "即使开关打开,一份没有基线来源的包裹也不允许推送(baseline_source='none' 时" "delta 无意义,卡片只能报存量,而存量不是这次转岗造成的)。" "allowed/push_baseline_sources 决定哪些基线来源算数 —— peer(同侪分布)是统计" "信号不是政策,默认不解锁推送。admin_digest_enabled 控制 IT 侧汇总(不打扰员工)。" ) def _has_table(conn, name: str) -> bool: return sa.inspect(conn).has_table(name) def _has_column(conn, table: str, column: str) -> bool: if not _has_table(conn, table): return False return any(c["name"] == column for c in sa.inspect(conn).get_columns(table)) def upgrade() -> None: conn = op.get_bind() # ── 1. suppression columns ────────────────────────────────────── if _has_table(conn, REVIEWS): if not _has_column(conn, REVIEWS, "notify_suppressed_at"): op.add_column( REVIEWS, sa.Column("notify_suppressed_at", sa.DateTime(timezone=True)), ) if not _has_column(conn, REVIEWS, "notify_suppressed_reason"): op.add_column( REVIEWS, sa.Column("notify_suppressed_reason", sa.String(64)), ) # ── the delta columns ─────────────────────────────────────── # # Added here rather than with the engine that fills them, so the # gate is complete from the moment it exists: the send path asks # ``settings.may_push(baseline_source, delta_items)``, and every # existing row answers ``('none', 0)`` — meaning no packet built # before the engine can ever be pushed, whatever an operator # switches on. Turning that invariant on two PRs later would # leave a window in which flipping one boolean re-creates the # 2026-09-05 cards. # # NOT NULL with a server default: a nullable "we don't know" # would have to be interpreted at every read, and the safe # interpretation is exactly what the default already says. if not _has_column(conn, REVIEWS, "baseline_source"): op.add_column( REVIEWS, sa.Column( "baseline_source", sa.String(16), nullable=False, server_default="none", ), ) if not _has_column(conn, REVIEWS, "delta_items"): op.add_column( REVIEWS, sa.Column( "delta_items", sa.Integer, nullable=False, server_default="0", ), ) # ── 2. stamp the backlog ──────────────────────────────────── # Every packet that has not been notified yet. Deliberately NOT # restricted to ``status='open'``: a superseded packet that was # never sent must stay un-sendable too, and a status transition # is not supposed to resurrect a withheld notification. conn.execute( sa.text( f"UPDATE {REVIEWS} " "SET notify_suppressed_at = NOW(), " " notify_suppressed_reason = :reason " "WHERE notified_at IS NULL " " AND notify_suppressed_at IS NULL", ), {"reason": SUPPRESS_REASON}, ) # ── 3. turn the two push crons off ────────────────────────────── # UPDATE only. A row that does not exist is a deployment that never # ran migration 236 / 568, and inserting a disabled row there would # be inventing state. if _has_table(conn, JOBS): conn.execute( sa.text( f"UPDATE {JOBS} SET enabled = 0, updated_at = NOW() " "WHERE handler_key IN (:a, :b)", ), {"a": NOTIFY_JOB, "b": SWEEP_JOB}, ) # ── 4. the policy row ─────────────────────────────────────────── if _has_table(conn, SETTINGS): conn.execute( sa.text( f"INSERT INTO {SETTINGS} (`key`, `value`, description, " "created_at, updated_at) " "SELECT :key, :value, :desc, NOW(), NOW() FROM DUAL " f"WHERE NOT EXISTS (SELECT 1 FROM {SETTINGS} WHERE `key` = :key)", ), { "key": SETTING_KEY, "value": json.dumps(_SETTING_VALUE), "desc": _SETTING_DESC, }, ) def downgrade() -> None: conn = op.get_bind() if _has_table(conn, SETTINGS): conn.execute( sa.text(f"DELETE FROM {SETTINGS} WHERE `key` = :key"), {"key": SETTING_KEY}, ) # The crons are NOT re-enabled on downgrade. Rolling the schema back # is not a statement that the baseline problem was solved, and a # downgrade that silently starts pushing to managers again is the # worst possible surprise. if _has_column(conn, REVIEWS, "delta_items"): op.drop_column(REVIEWS, "delta_items") if _has_column(conn, REVIEWS, "baseline_source"): op.drop_column(REVIEWS, "baseline_source") if _has_column(conn, REVIEWS, "notify_suppressed_reason"): op.drop_column(REVIEWS, "notify_suppressed_reason") if _has_column(conn, REVIEWS, "notify_suppressed_at"): op.drop_column(REVIEWS, "notify_suppressed_at")