"""Regression: migration 472 must apply on prod TiDB, and must resume after its own half-applied first attempt. 2026-08-16 production deploy, two failures stacked on one another: 1. The three columns FK'ing to ``identity_persons`` were declared ``sa.CHAR(36)`` with no collation, so they inherited the *server* default (``utf8mb4_bin`` on prod TiDB) while the parent is ``utf8mb4_unicode_ci``. TiDB refused ``fk_2`` with error 3780. Local MySQL 8 has one collation across both and cannot reproduce it — the same blind spot as migrations 126 and 437. 2. TiDB DDL is non-transactional, so the tables created before that failure stayed behind while ``alembic_version`` remained at 471. ``intelligence_snapshots`` was left in the schema, empty, carrying the *binary* server default. Fixing only (1) is not enough, and that is the point of this file. A plain "skip the table if it already exists" guard leaves the stale binary ``intelligence_snapshots`` in place, and the very next statement creates ``intelligence_analyses`` with the *unicode* table charset and a ``snapshot_id`` FK into it — 3780 again, now on ``intelligence_analyses_ibfk_1`` instead of ``fk_2``. Verified by mutation: neutralising ``_reset_if_stale`` in the migration reproduces exactly that error on a prod-shaped scratch schema. So the migration drops leftovers whose collation disagrees, and these tests pin both halves: that it resumes, and that resuming produces a schema whose FK columns match their parents. """ from __future__ import annotations import importlib.util import os import pathlib import uuid import pytest import sqlalchemy as sa from alembic.migration import MigrationContext from alembic.operations import Operations from ..conftest import _TEST_SYNC_URL _MIGRATION_472 = ( pathlib.Path(__file__).resolve().parents[2] / "alembic" / "versions" / "472_intelligence_kernel.py" ) # Prod TiDB's split: identity_persons is unicode_ci, the server default # (and therefore anything created without an explicit charset) is binary. _BIN = "utf8mb4_bin" _CI = "utf8mb4_unicode_ci" _STALE_SNAPSHOTS_DDL = """ CREATE TABLE intelligence_snapshots ( id CHAR(36) NOT NULL, subject_type VARCHAR(24) NOT NULL, subject_id VARCHAR(128) NOT NULL, audience_key VARCHAR(128) NOT NULL, score_version VARCHAR(32) NOT NULL, facts_hash VARCHAR(64) NOT NULL, overall_score DECIMAL(5,2) NOT NULL, confidence_score DECIMAL(5,2) NOT NULL, grade VARCHAR(24) NOT NULL, dimensions JSON NOT NULL, aggregate_facts JSON NOT NULL, reference_map JSON NOT NULL, data_as_of DATETIME NOT NULL, created_at DATETIME NOT NULL DEFAULT NOW(), PRIMARY KEY (id) ) CHARSET=utf8mb4 COLLATE={collation} """ def _load_migration(): spec = importlib.util.spec_from_file_location( "m472_prodshape", _MIGRATION_472, ) assert spec and spec.loader module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module def _column_collation(conn, table: str, column: str) -> str | None: return conn.execute( sa.text( "SELECT COLLATION_NAME FROM information_schema.COLUMNS " "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :t " "AND COLUMN_NAME = :c" ), {"t": table, "c": column}, ).scalar() class _ProdShapedSchema: """A scratch schema with prod TiDB's collations, dropped on exit. Needs a connection allowed to create a schema: CI supplies ``TEST_DATABASE_ADMIN_URL`` (root), otherwise the ordinary test user is tried, which locally is unprivileged. Callers skip when neither works — a privilege difference is not a code defect. """ def __init__(self) -> None: self.name = f"helix_m472_{uuid.uuid4().hex[:8]}" admin_url = os.environ.get("TEST_DATABASE_ADMIN_URL") self.base = ( admin_url.rstrip("/").rpartition("/")[0] if admin_url else _TEST_SYNC_URL.rpartition("/")[0] ) self.admin = sa.create_engine( self.base + "/", isolation_level="AUTOCOMMIT", ) self.engine: sa.Engine | None = None def __enter__(self): try: with self.admin.connect() as c: c.execute(sa.text( f"CREATE DATABASE {self.name} CHARACTER SET utf8mb4" )) except sa.exc.SQLAlchemyError as exc: self.admin.dispose() pytest.skip(f"cannot create a scratch schema: {exc}") self.engine = sa.create_engine( f"{self.base}/{self.name}", isolation_level="AUTOCOMMIT", ) conn = self.engine.connect() self._conn = conn conn.execute(sa.text( "CREATE TABLE identity_persons " "(id CHAR(36) NOT NULL, PRIMARY KEY (id)) " f"CHARSET=utf8mb4 COLLATE={_CI}" )) # 472 seeds its release gate into platform_settings. conn.execute(sa.text( "CREATE TABLE platform_settings (" " `key` VARCHAR(191) NOT NULL, value TEXT, description TEXT," " created_at DATETIME NOT NULL, updated_at DATETIME NOT NULL," " PRIMARY KEY (`key`)" f") CHARSET=utf8mb4 COLLATE={_BIN}" )) return conn def __exit__(self, *exc) -> None: self._conn.close() if self.engine is not None: self.engine.dispose() with self.admin.connect() as c: c.execute(sa.text(f"DROP DATABASE IF EXISTS {self.name}")) self.admin.dispose() def _run_upgrade(conn) -> None: ctx = MigrationContext.configure(conn) with Operations.context(ctx): _load_migration().upgrade() def _assert_healthy(conn) -> None: tables = [ r[0] for r in conn.execute(sa.text( "SELECT TABLE_NAME FROM information_schema.TABLES " "WHERE TABLE_SCHEMA = DATABASE() " "AND TABLE_NAME LIKE 'intelligence%' ORDER BY TABLE_NAME" )) ] assert tables == [ "intelligence_analyses", "intelligence_generation_leases", "intelligence_recommendations", "intelligence_snapshots", ], tables # Every person FK, and every intra-migration FK, matches its parent. for table, column in ( ("intelligence_analyses", "generated_by_user_id"), ("intelligence_recommendations", "assigned_to_person_id"), ("intelligence_recommendations", "last_changed_by_user_id"), ("intelligence_analyses", "snapshot_id"), ("intelligence_recommendations", "snapshot_id"), ("intelligence_recommendations", "analysis_id"), ): assert _column_collation(conn, table, column) == _CI, (table, column) fks = conn.execute(sa.text( "SELECT COUNT(*) FROM information_schema.KEY_COLUMN_USAGE " "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME LIKE 'intelligence%' " "AND REFERENCED_TABLE_NAME IS NOT NULL" )).scalar() # 3 into identity_persons + 3 between the new tables. A silently # dropped FK would leave analyses/recommendations unanchored. assert fks == 6, fks assert conn.execute(sa.text( "SELECT COUNT(*) FROM platform_settings WHERE `key` = :k" ), {"k": "intelligence.enabled"}).scalar() == 1 def test_applies_on_a_clean_prod_shaped_schema() -> None: """The original failure: person FKs must not inherit the binary default. Before the fix this raises 3780 on ``fk_2``, which is how the 2026-08-16 deploy died with the chain stuck at 471. """ with _ProdShapedSchema() as conn: _run_upgrade(conn) _assert_healthy(conn) def test_resumes_over_a_stale_binary_leftover() -> None: """The second failure, the one a bare existence guard would not fix. Reproduces prod exactly: ``intelligence_snapshots`` already present, empty, ``utf8mb4_bin``. Skipping it and creating its children at ``utf8mb4_unicode_ci`` re-raises 3780 on ``snapshot_id``; the migration must rebuild it instead. """ with _ProdShapedSchema() as conn: conn.execute(sa.text(_STALE_SNAPSHOTS_DDL.format(collation=_BIN))) assert _column_collation(conn, "intelligence_snapshots", "id") == _BIN _run_upgrade(conn) assert _column_collation(conn, "intelligence_snapshots", "id") == _CI _assert_healthy(conn) def test_resumes_over_a_leftover_that_is_already_correct() -> None: """Not every leftover is stale — a matching one must survive untouched. Dropping and rebuilding unconditionally would be the easy fix and the wrong one: it would discard rows on any environment where the earlier run got far enough to be usable. Here the leftover already collates correctly, so the migration must keep it and only build what's missing. """ with _ProdShapedSchema() as conn: conn.execute(sa.text(_STALE_SNAPSHOTS_DDL.format(collation=_CI))) conn.execute(sa.text( "INSERT INTO intelligence_snapshots (id, subject_type, " "subject_id, audience_key, score_version, facts_hash, " "overall_score, confidence_score, grade, dimensions, " "aggregate_facts, reference_map, data_as_of) VALUES " "('keep-me', 'person', 'p1', 'a', 'v1', 'h', 1, 1, 'g', " "'{}', '{}', '{}', NOW())" )) _run_upgrade(conn) assert conn.execute(sa.text( "SELECT COUNT(*) FROM intelligence_snapshots WHERE id = 'keep-me'" )).scalar() == 1 _assert_healthy(conn) def test_refuses_to_drop_a_stale_table_holding_rows() -> None: """The premise of the drop, checked rather than trusted. The migration may drop leftovers only because ``upgrade()`` runs solely while the version is below 472, so a leftover can only be debris from a failed run — and debris is empty. If that premise is ever false, stopping loudly beats losing rows silently. """ with _ProdShapedSchema() as conn: conn.execute(sa.text(_STALE_SNAPSHOTS_DDL.format(collation=_BIN))) conn.execute(sa.text( "INSERT INTO intelligence_snapshots (id, subject_type, " "subject_id, audience_key, score_version, facts_hash, " "overall_score, confidence_score, grade, dimensions, " "aggregate_facts, reference_map, data_as_of) VALUES " "('surprise', 'person', 'p1', 'a', 'v1', 'h', 1, 1, 'g', " "'{}', '{}', '{}', NOW())" )) with pytest.raises(RuntimeError, match="refusing to drop"): _run_upgrade(conn) assert conn.execute(sa.text( "SELECT COUNT(*) FROM intelligence_snapshots" )).scalar() == 1 def test_rerunning_a_completed_upgrade_is_a_no_op() -> None: """Idempotence over the whole migration, not just the first table. Deploys get retried. The guards exist so a retry lands on the same schema rather than "table already exists" — which, once the chain is stuck, blocks every migration after this one too. """ with _ProdShapedSchema() as conn: _run_upgrade(conn) _run_upgrade(conn) _assert_healthy(conn)