"""自证的 AI 建议 + 「没人回答」不再是终点(访问对象闭环 PR5)。 半年度访问自查(`self_attestation`)有两个洞,这一条迁移给两个都铺路: **① 持有人没有可以据以判断的东西。** 一份资深工程师的清单是几十行 `jira-software-users` / `analytics-prod.events_raw` 这样的名字;冷着问,唯一 安全的答案是「都留着」,而一次人人全留的复核**产出了证据、什么也没改变**。 PR4 已经把缺的那半做出来了(`asset_seat_usage_signals` 知道每个人最后一次 真的用某个产品是什么时候)。新表 `auditor_self_attestation_advice` 存放按 (campaign × 持有人) 生成的建议:一段摘要 + 每个访问类别一个 keep/release 初步建议 + 生成建议时用到的事实快照。**建议落在人的决定旁边,永不落进决定里。** **② `no_response` 是个耸肩。** `complete_certification` 在收尾时把没人回答的 条目一律盖成 `no_response`(`decided_by_person_id` 保持 NULL),**权限原样留着**。 于是这个流程为「做过一次复核」产出了 SOC2 证据,而一个从不回答的人什么都没丢, 半年后同一份清单再来一遍。新的 campaign_type `no_response_followup` 让那些条目 自动开一个**补审**,reviewer 是持有人的直属经理 —— 走的是既有的 `decide_item` + 回收流水线,不另造一套。**没有任何东西被自动撤销**:补审只是把决定权从一个 没回答的人手里,交到一个能回答的人手里。 两个平台设置: * `access_review.self_attestation.ai_advice_enabled` —— AI 建议开关,**默认关** (与 2026-09-05 决定①「新增员工触达逐类灰度、默认关」一致)。 * `access_review.self_attestation.no_response_policy` —— `report_only`(默认, 保持今天的行为)/ `escalate_manager`(开补审 campaign)。 一条 cron `auditor.self_attestation_advice`:每天 01:30 UTC 给开着的自证 campaign 分批生成建议。分批是因为一次 campaign 有几百个持有人,每人一次模型 调用 —— 放在开 campaign 的那个请求里会超时。 无回填:建议是模型产物,历史 campaign 无从追溯,也不该追溯。 """ from __future__ import annotations import json import uuid import sqlalchemy as sa from alembic import op from app.core.migration_helpers import id_type_for, table_kw_for revision = "621" # Both ``revision`` AND ``down_revision`` are settled only at push. # This file has been renumbered SEVEN times in one day — 607 taken by a # third party, 608 by the asset session, 613 by #3462, 617 by #3467 — # and its parent moved five times more (614 → 616 → 617 → 618 → 619 → 620). The 617 # collision was the worst shape: a duplicate revision id is not two heads, # it is one id meaning two different things, and alembic refuses outright. # 619 then went the same way (#3470), and 620 after it (#3476), both # taken while this branch sat in review — which is why the number is re-read from `origin/main` and # never carried over from the last verification run. # Numeric gaps are irrelevant to alembic — ``down_revision`` is the ONLY # thing that decides the chain, so a file numbered 619 whose parent is 614 # sits on a second head the moment 615/616 exist. Re-checked against # ``origin/main`` immediately before push, which is the only moment either # value is actually settled. down_revision = "620" branch_labels = None depends_on = None TABLE = "auditor_self_attestation_advice" JOBS = "scheduled_jobs" SETTINGS = "platform_settings" JOB_KEY = "auditor.self_attestation_advice" KEY_AI_ENABLED = "access_review.self_attestation.ai_advice_enabled" KEY_NO_RESPONSE = "access_review.self_attestation.no_response_policy" _JOB_DESC = ( "每天 01:30 UTC。给当前开着的半年度访问自查 campaign 分批生成 AI 建议" "(每个持有人一行 auditor_self_attestation_advice):一段摘要 + 每个访问类别一个 " "keep/release 初步建议。事实来自 campaign 自己的条目 + PR4 的使用信号," "模型不能引用事实里没有的数字、不能发明类别、也不能把「无法确认使用情况」" "说成「闲置」(代码级校验,不只是提示词约束)。建议落在持有人的决定**旁边**," "永不落进决定里。开关 access_review.self_attestation.ai_advice_enabled 默认关。" ) def _has_table(bind, name: str) -> bool: return bool(bind.execute( sa.text( "SELECT 1 FROM information_schema.tables " "WHERE table_schema = DATABASE() AND table_name = :t LIMIT 1", ), {"t": name}, ).scalar()) def upgrade() -> None: bind = op.get_bind() if not _has_table(bind, TABLE): # ``id_type_for`` / ``table_kw_for``, NOT a hand-rolled read of # ``information_schema.columns.column_type``. # # Migration 612 wedged the prod chain by reaching for # ``sa.Uuid()`` (CHAR(32) on MySQL) against a CHAR(36) parent. # The obvious-looking fix — read the parent's ``column_type`` and # echo it — is ALSO wrong, and this migration shipped that # version briefly: the type string carries no collation, and # errno 3780 fires on a collation mismatch even when the data # types are identical. A column with no explicit ``COLLATE`` # inherits the TABLE default, so the FK only holds while the new # table's default happens to match the parent's. # # It passed on a scratch database created in one go (uniform # collation) and failed on a database built through the real # chain — which is the same asymmetry ``id_type_for``'s docstring # describes for prod TiDB vs local MySQL 8. That helper already # encodes the whole lesson (migration 126, and the 2026-07-29 # deploy break), so the right move is to ask it rather than to # re-derive it here for the third time. cert_id = id_type_for("access_certifications") person_id = id_type_for("identity_persons") kw = table_kw_for("access_certifications") op.create_table( TABLE, sa.Column("id", cert_id, primary_key=True), sa.Column("certification_id", cert_id, nullable=False), # No FK by design (the TiDB-FK-safe convention this domain's # person_id columns follow), but the type still matches the # parent so a future FK is not blocked. sa.Column("person_id", person_id, nullable=False), sa.Column("summary_zh", sa.Text(), nullable=True), sa.Column("summary_en", sa.Text(), nullable=True), sa.Column("type_recommendations", sa.JSON(), nullable=True), sa.Column("facts_snapshot", sa.JSON(), nullable=True), sa.Column("failure_code", sa.String(64), nullable=True), sa.Column("generated_at", sa.DateTime(), nullable=True), sa.Column( "created_at", sa.DateTime(), nullable=False, server_default=sa.text("CURRENT_TIMESTAMP"), ), sa.Column( "updated_at", sa.DateTime(), nullable=False, server_default=sa.text("CURRENT_TIMESTAMP"), server_onupdate=sa.text("CURRENT_TIMESTAMP"), ), sa.ForeignKeyConstraint( ["certification_id"], ["access_certifications.id"], name="fk_self_att_advice_cert", ondelete="CASCADE", ), sa.UniqueConstraint( "certification_id", "person_id", name="uq_self_att_advice_cert_person", ), **kw, ) op.create_index("ix_self_att_advice_cert", TABLE, ["certification_id"]) if _has_table(bind, SETTINGS): # ``value`` is a JSON column — the payload must be json.dumps'd, # not a bare string (a raw ``report_only`` is invalid JSON and the # INSERT fails outright). Same shape as migration 029's seed. for key, value, desc in ( (KEY_AI_ENABLED, False, "半年度访问自查是否给持有人展示 AI 初步建议。默认关 —— 新增的员工" "触达逐类灰度(2026-09-05 决定①)。建议永不写进持有人的决定。"), (KEY_NO_RESPONSE, "report_only", "自证收尾时对「没人回答」的条目做什么。report_only=保持今天的行为" "(只盖 no_response + 合规事件);escalate_manager=自动开一个 " "no_response_followup 补审 campaign,reviewer 是持有人的直属经理," "走既有 decide_item + 回收流水线。任何一档都不会自动撤销权限。"), ): bind.execute( sa.text( f"INSERT INTO {SETTINGS} (`key`, value, description, " " created_at, updated_at) " "SELECT :k, :v, :d, NOW(), NOW() FROM DUAL " f"WHERE NOT EXISTS (SELECT 1 FROM {SETTINGS} WHERE `key` = :k)", ), {"k": key, "v": json.dumps(value), "d": desc}, ) if _has_table(bind, JOBS): bind.execute( sa.text( f"INSERT INTO {JOBS} " "(id, name, display_name, description, handler_key, " " trigger_type, cron_expr, timezone, enabled, config, " " run_count, failure_count, created_at, updated_at) " "SELECT :id, :name, :display, :desc, :handler, 'cron', " ":cron, 'UTC', 1, :config, 0, 0, NOW(), NOW() " "FROM DUAL WHERE NOT EXISTS " f"(SELECT 1 FROM {JOBS} WHERE handler_key = :handler)", ), { "id": str(uuid.uuid4()), "name": JOB_KEY, "display": "访问自查 · AI 建议生成", "desc": _JOB_DESC, "handler": JOB_KEY, "cron": "30 1 * * *", "config": "{}", }, ) def downgrade() -> None: bind = op.get_bind() if _has_table(bind, JOBS): bind.execute( sa.text(f"DELETE FROM {JOBS} WHERE handler_key = :h"), {"h": JOB_KEY}, ) if _has_table(bind, SETTINGS): for key in (KEY_AI_ENABLED, KEY_NO_RESPONSE): bind.execute( sa.text(f"DELETE FROM {SETTINGS} WHERE `key` = :k"), {"k": key}, ) if _has_table(bind, TABLE): op.drop_table(TABLE)