"""Cross-domain Workflow Journal + Leaver Blocked hard-gate + Relationship graph. Revision ID: 091 Revises: 090 Create Date: 2026-04-30 Three coupled feature lines that ship together: 1. Workflow Journal (`workflow_runs` + `workflow_step_events`) — cross-domain progress event recorder so the employee-facing /me page and the Leaver Blocked dashboard render a single timeline regardless of which domain emitted the event. 2. Leaver Blocked hard-gate (`leaver_block_audit` + `leaver_dual_sign_overrides` + columns on `offboarding_processes` and `identity_persons`) — make the ``employment_status='left'`` invariant impossible to bypass. 3. Relationship graph (`rel_relationships`) — materialised edges across every owner/member/responsibility relation Helix tracks. This is NOT Temporal — Helix runs on a single TiDB cluster by hard policy, and the journal sits in those same tables managed by Alembic. ═══════════════════════════════════════════════════════════════════════ TiDB FK strictness ═══════════════════════════════════════════════════════════════════════ Every FK column is declared via ``_fk_source_col_type`` (same helper shape as migration 088) which reads ``information_schema.COLUMNS`` for the parent's actual DATA_TYPE + COLLATION_NAME and builds a child column type that matches exactly. TiDB rejects the FK otherwise with errno 3780 ("Referencing column ... and referenced column ... in foreign key constraint ... are incompatible") even when the declared types look identical at the SQLAlchemy level. ═══════════════════════════════════════════════════════════════════════ Idempotency ═══════════════════════════════════════════════════════════════════════ Every CREATE TABLE / ADD COLUMN / INSERT INTO scheduled_jobs is guarded with an existence check so re-running ``alembic upgrade head`` is safe. Downgrade drops the new tables/columns and seeds — only used for local round-trips. """ from __future__ import annotations import json import logging import uuid import sqlalchemy as sa from alembic import op logger = logging.getLogger(__name__) revision = "091" down_revision = "090" branch_labels = None depends_on = None # ── Idempotency helpers ────────────────────────────────────────────── def _table_exists(bind, name: str) -> bool: return bind.execute( sa.text( "SELECT 1 FROM information_schema.TABLES " "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :n" ), {"n": name}, ).scalar() is not None def _column_exists(bind, table: str, column: str) -> bool: return bind.execute( sa.text( "SELECT 1 FROM information_schema.COLUMNS " "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :t " " AND COLUMN_NAME = :c" ), {"t": table, "c": column}, ).scalar() is not None def _fk_source_col_type( bind, target_table: str, target_column: str, *, fallback_type: str = "char", length: int = 36, ): """Return a SQLAlchemy type matching the parent column's actual DATA_TYPE + COLLATION_NAME, so TiDB accepts the resulting FK. Lift directly from migration 088 — same shape, same intent. """ row = bind.execute( sa.text( "SELECT DATA_TYPE, COLLATION_NAME FROM " "information_schema.COLUMNS " "WHERE TABLE_SCHEMA = DATABASE() " " AND TABLE_NAME = :t AND COLUMN_NAME = :c" ), {"t": target_table, "c": target_column}, ).fetchone() if row: data_type = (row[0] or fallback_type).lower() collation = row[1] else: data_type = fallback_type.lower() collation = None type_cls = sa.CHAR if data_type == "char" else sa.String if collation: return type_cls(length, collation=collation) return type_cls(length) _IDEMPOTENT_RETRY_HINTS = ( "duplicate key name", "duplicate entry", "duplicate foreign key", "errno: 121", "check that column/key exists", ) def _try_swallow(label: str, fn) -> None: try: fn() except Exception as exc: # noqa: BLE001 msg = str(exc).lower() if any(hint in msg for hint in _IDEMPOTENT_RETRY_HINTS): logger.warning( "migration 091: %s skipped (idempotent retry): %s", label, exc, ) return logger.error( "migration 091: %s failed with non-idempotent error: %s", label, exc, ) raise # ── upgrade ───────────────────────────────────────────────────────── def upgrade() -> None: bind = op.get_bind() # FK target column types — looked up at runtime so collation / # exact data_type match the parent. Must be done BEFORE we create # any child table. persons_id = _fk_source_col_type(bind, "identity_persons", "id") offboarding_proc_id = _fk_source_col_type( bind, "offboarding_processes", "id", ) # ════════════════════════════════════════════════════════════ # 1. Workflow Journal # ════════════════════════════════════════════════════════════ if not _table_exists(bind, "workflow_runs"): op.create_table( "workflow_runs", sa.Column("id", sa.CHAR(36), primary_key=True), sa.Column("workflow_type", sa.String(64), nullable=False), sa.Column("person_id", persons_id, nullable=True), sa.Column("related_resource_type", sa.String(64), nullable=True), sa.Column("related_resource_id", sa.CHAR(36), nullable=True), sa.Column( "status", sa.String(32), nullable=False, server_default=sa.text("'running'"), ), sa.Column("current_step", sa.String(128), nullable=True), sa.Column("step_count", sa.Integer, nullable=False, server_default=sa.text("0")), sa.Column("step_log", sa.JSON, nullable=True), sa.Column("context", sa.JSON, nullable=True), sa.Column("trace_id", sa.CHAR(36), nullable=False), sa.Column("trigger_source", sa.String(64), nullable=False), sa.Column("triggered_by_user_id", persons_id, nullable=True), sa.Column("started_at", sa.DateTime(timezone=True), nullable=False), sa.Column("last_step_at", sa.DateTime(timezone=True), nullable=True), sa.Column("completed_at", sa.DateTime(timezone=True), nullable=True), sa.Column("error_class", sa.String(128), nullable=True), sa.Column("error_message", sa.Text, nullable=True), sa.Column( "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.Column( "updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), server_onupdate=sa.func.now(), ), sa.ForeignKeyConstraint( ["person_id"], ["identity_persons.id"], name="fk_workflow_runs_person", ), sa.ForeignKeyConstraint( ["triggered_by_user_id"], ["identity_persons.id"], name="fk_workflow_runs_triggered_by", ), sa.Index("idx_wf_runs_person_started", "person_id", "started_at"), sa.Index("idx_wf_runs_type_status", "workflow_type", "status"), sa.Index("idx_wf_runs_related", "related_resource_type", "related_resource_id"), sa.Index("idx_wf_runs_status", "status"), ) if not _table_exists(bind, "workflow_step_events"): # workflow_runs.id is plain CHAR(36) (just created) so a direct # CHAR(36) here matches; collation defaults are uniform within # this migration's freshly-created tables. op.create_table( "workflow_step_events", sa.Column("id", sa.CHAR(36), primary_key=True), sa.Column("workflow_run_id", sa.CHAR(36), nullable=False), sa.Column("sequence", sa.Integer, nullable=False), sa.Column("step_name", sa.String(128), nullable=False), sa.Column("status", sa.String(32), nullable=False), sa.Column("started_at", sa.DateTime(timezone=True), nullable=False), sa.Column("completed_at", sa.DateTime(timezone=True), nullable=True), sa.Column("duration_ms", sa.Integer, nullable=True), sa.Column("attempt", sa.Integer, nullable=False, server_default=sa.text("1")), sa.Column("actor_user_id", persons_id, nullable=True), sa.Column( "actor_kind", sa.String(32), nullable=False, server_default=sa.text("'system'"), ), sa.Column("error_class", sa.String(128), nullable=True), sa.Column("error_message", sa.Text, nullable=True), sa.Column("payload", sa.JSON, nullable=True), sa.Column( "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.ForeignKeyConstraint( ["workflow_run_id"], ["workflow_runs.id"], ondelete="CASCADE", name="fk_wf_step_events_run", ), sa.ForeignKeyConstraint( ["actor_user_id"], ["identity_persons.id"], name="fk_wf_step_events_actor", ), sa.Index("idx_wf_step_events_run_seq", "workflow_run_id", "sequence"), sa.Index("idx_wf_step_events_step_started", "step_name", "started_at"), ) # ════════════════════════════════════════════════════════════ # 2. Leaver Blocked hard-gate # ════════════════════════════════════════════════════════════ if not _table_exists(bind, "leaver_block_audit"): op.create_table( "leaver_block_audit", sa.Column("id", sa.CHAR(36), primary_key=True), sa.Column("person_id", persons_id, nullable=False), sa.Column("attempted_at", sa.DateTime(timezone=True), nullable=False), sa.Column("attempted_by_user_id", persons_id, nullable=True), sa.Column("attempted_via", sa.String(64), nullable=False), sa.Column("target_status", sa.String(32), nullable=False), sa.Column("denied_reason", sa.String(64), nullable=False), sa.Column("pending_blocker_count", sa.Integer, nullable=False, server_default=sa.text("0")), sa.Column("pending_blockers", sa.JSON, nullable=False), sa.Column("detail", sa.JSON, nullable=True), sa.Column("trace_id", sa.CHAR(36), nullable=False), sa.Column("client_ip", sa.String(64), nullable=True), sa.Column("user_agent", sa.String(512), nullable=True), sa.Column( "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.ForeignKeyConstraint( ["person_id"], ["identity_persons.id"], name="fk_leaver_block_audit_person", ), sa.ForeignKeyConstraint( ["attempted_by_user_id"], ["identity_persons.id"], name="fk_leaver_block_audit_actor", ), sa.Index("idx_leaver_block_audit_person_at", "person_id", "attempted_at"), sa.Index("idx_leaver_block_audit_reason", "denied_reason"), sa.Index("idx_leaver_block_audit_via", "attempted_via"), ) if not _table_exists(bind, "leaver_dual_sign_overrides"): op.create_table( "leaver_dual_sign_overrides", sa.Column("id", sa.CHAR(36), primary_key=True), sa.Column("process_id", offboarding_proc_id, nullable=False), sa.Column("person_id", persons_id, nullable=False), sa.Column( "status", sa.String(32), nullable=False, server_default=sa.text("'pending_secondary'"), ), sa.Column("primary_signer_user_id", persons_id, nullable=False), sa.Column("primary_signed_at", sa.DateTime(timezone=True), nullable=False), sa.Column("primary_reason", sa.Text, nullable=False), sa.Column("secondary_signer_user_id", persons_id, nullable=True), sa.Column("secondary_signed_at", sa.DateTime(timezone=True), nullable=True), sa.Column("secondary_reason", sa.Text, nullable=True), sa.Column("overridden_blocker_ids", sa.JSON, nullable=False), sa.Column("overridden_blockers_snapshot", sa.JSON, nullable=False), sa.Column("executed_at", sa.DateTime(timezone=True), nullable=True), sa.Column("rejected_at", sa.DateTime(timezone=True), nullable=True), sa.Column("rejected_reason", sa.Text, nullable=True), sa.Column("expires_at", sa.DateTime(timezone=True), nullable=True), sa.Column("trace_id", sa.CHAR(36), nullable=False), sa.Column( "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.Column( "updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), server_onupdate=sa.func.now(), ), sa.ForeignKeyConstraint( ["process_id"], ["offboarding_processes.id"], ondelete="CASCADE", name="fk_dual_sign_process", ), sa.ForeignKeyConstraint( ["person_id"], ["identity_persons.id"], name="fk_dual_sign_person", ), sa.ForeignKeyConstraint( ["primary_signer_user_id"], ["identity_persons.id"], name="fk_dual_sign_primary", ), sa.ForeignKeyConstraint( ["secondary_signer_user_id"], ["identity_persons.id"], name="fk_dual_sign_secondary", ), sa.Index("idx_dual_sign_process_status", "process_id", "status"), sa.Index("idx_dual_sign_status", "status"), ) # New columns on offboarding_processes if not _column_exists(bind, "offboarding_processes", "current_step"): op.add_column( "offboarding_processes", sa.Column("current_step", sa.String(128), nullable=True), ) if not _column_exists(bind, "offboarding_processes", "step_log"): op.add_column( "offboarding_processes", sa.Column("step_log", sa.JSON, nullable=True), ) if not _column_exists(bind, "offboarding_processes", "first_blocked_at"): op.add_column( "offboarding_processes", sa.Column("first_blocked_at", sa.DateTime(timezone=True), nullable=True), ) if not _column_exists(bind, "offboarding_processes", "sla_breach_level"): op.add_column( "offboarding_processes", sa.Column( "sla_breach_level", sa.String(16), nullable=False, server_default=sa.text("'none'"), ), ) if not _column_exists(bind, "offboarding_processes", "workflow_run_id"): op.add_column( "offboarding_processes", sa.Column("workflow_run_id", sa.CHAR(36), nullable=True), ) # New column on identity_persons if not _column_exists(bind, "identity_persons", "force_logout_at"): op.add_column( "identity_persons", sa.Column("force_logout_at", sa.DateTime(timezone=True), nullable=True), ) # ════════════════════════════════════════════════════════════ # 3. Relationship graph # ════════════════════════════════════════════════════════════ if not _table_exists(bind, "rel_relationships"): op.create_table( "rel_relationships", sa.Column("id", sa.CHAR(36), primary_key=True), sa.Column("from_type", sa.String(32), nullable=False), sa.Column("from_id", sa.String(64), nullable=False), sa.Column("from_label", sa.String(255), nullable=True), sa.Column("to_type", sa.String(32), nullable=False), sa.Column("to_id", sa.String(64), nullable=False), sa.Column("to_label", sa.String(255), nullable=True), sa.Column("rel_type", sa.String(48), nullable=False), sa.Column("attributes", sa.JSON, nullable=True), sa.Column( "source", sa.String(48), nullable=False, server_default=sa.text("'sync'"), ), sa.Column("weight", sa.Float, nullable=False, server_default=sa.text("1.0")), sa.Column("first_seen_at", sa.DateTime(timezone=True), nullable=False), sa.Column("last_seen_at", sa.DateTime(timezone=True), nullable=False), sa.Column("deleted_at", sa.DateTime(timezone=True), nullable=True), sa.Column( "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.Column( "updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), server_onupdate=sa.func.now(), ), sa.UniqueConstraint( "from_type", "from_id", "to_type", "to_id", "rel_type", name="uq_rel_edge", ), sa.Index("idx_rel_from", "from_type", "from_id"), sa.Index("idx_rel_to", "to_type", "to_id"), sa.Index("idx_rel_type", "rel_type"), sa.Index("idx_rel_last_seen", "last_seen_at"), ) # ════════════════════════════════════════════════════════════ # 4. Seed scheduled_jobs rows (only ones whose handler ships in # this PR; security_* jobs land with the security module) # ════════════════════════════════════════════════════════════ seeds = [ { "name": "relationships.full_sync", "display": "关系图谱全量同步", "desc": ( "每日 04:30 全量重建 rel_relationships:扫描 identity_persons 经理链、" "asset_items 持有人、software_seat_allocations、cmdb_servers / cmdb_domains / " "cmdb_cloud_scopes / cmdb_cloud_access_grants、email_group_owners、" "asset_software_catalog 业务责任人 / IT 联系人、integrations 创建者," "把所有 owner / member / responsible 关系归一到一张图谱表。" ), "trigger": "cron", "cron": "30 4 * * *", "interval": None, "config": {}, }, { "name": "relationships.incremental_sync", "display": "关系图谱增量同步", "desc": ( "每 30 分钟扫一次最近变更的 owner FK,把增量边推进 rel_relationships。" "全量同步给基线,增量同步保证图谱与现实只差几十分钟。" ), "trigger": "interval", "cron": None, "interval": 1800, "config": {}, }, { "name": "leaver.escalation_tick", "display": "Leaver Blocked SLA 升级 tick", "desc": ( "每 30 分钟检查 first_blocked_at 已超 24h/48h/72h 的离职人," "依次 DM manager / DM IT admin / 创 critical compliance_incident," "更新 sla_breach_level,幂等于 reminder_log。" ), "trigger": "interval", "cron": None, "interval": 1800, "config": {}, }, ] if _table_exists(bind, "scheduled_jobs"): for seed in seeds: params: dict = { "id": str(uuid.uuid4()), "name": seed["name"], "display": seed["display"], "desc": seed["desc"], "handler": seed["name"], "trigger": seed["trigger"], "cron": seed["cron"], "interval": seed["interval"], "config": json.dumps(seed["config"]), } bind.execute( sa.text( "INSERT INTO scheduled_jobs " "(id, name, display_name, description, handler_key, " "trigger_type, cron_expr, interval_seconds, " "timezone, enabled, config, run_count, failure_count, " "created_at, updated_at) " "SELECT :id, :name, :display, :desc, :handler, :trigger, " ":cron, :interval, 'Asia/Shanghai', 1, :config, 0, 0, " "NOW(), NOW() FROM DUAL " "WHERE NOT EXISTS (SELECT 1 FROM scheduled_jobs WHERE name = :name)" ), params, ) # ── downgrade ─────────────────────────────────────────────────────── def downgrade() -> None: """Best-effort drop. Production never downgrades.""" bind = op.get_bind() seed_names = [ "relationships.full_sync", "relationships.incremental_sync", "leaver.escalation_tick", ] if _table_exists(bind, "scheduled_jobs"): for name in seed_names: bind.execute( sa.text("DELETE FROM scheduled_jobs WHERE name = :n"), {"n": name}, ) for tbl in ( "rel_relationships", "leaver_dual_sign_overrides", "leaver_block_audit", "workflow_step_events", "workflow_runs", ): if _table_exists(bind, tbl): op.drop_table(tbl) if _column_exists(bind, "identity_persons", "force_logout_at"): op.drop_column("identity_persons", "force_logout_at") for col in ( "workflow_run_id", "sla_breach_level", "first_blocked_at", "step_log", "current_step", ): if _column_exists(bind, "offboarding_processes", col): op.drop_column("offboarding_processes", col)