Reject duplicate module migration revisions

src/govoplan_core/db/migrations.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import ast
 from collections.abc import Iterable, Mapping
 from dataclasses import dataclass, replace
 import json
@@ -574,9 +575,73 @@ def alembic_config(
         config.attributes["enabled_modules"] = tuple(enabled_modules)
     if manifest_factories:
         config.attributes["manifest_factories"] = tuple(manifest_factories)
+    validate_unique_migration_revisions(config)
     return config
 
 
+def validate_unique_migration_revisions(config: Config) -> None:
+    """Reject duplicate revision IDs before Alembic assembles the shared graph.
+
+    Module migrations use separate version directories, but Alembic revision IDs
+    still occupy one global namespace. Alembic can otherwise resolve a duplicate
+    to the wrong module and report a misleading ancestor/head overlap.
+    """
+
+    locations = tuple(
+        Path(value).resolve()
+        for value in config.get_main_option("version_locations", "").split(os.pathsep)
+        if value.strip()
+    )
+    owners: dict[str, list[Path]] = {}
+    for location in locations:
+        if not location.is_dir():
+            continue
+        for path in sorted(location.glob("*.py")):
+            revision = _literal_migration_revision(path)
+            if revision:
+                owners.setdefault(revision, []).append(path)
+
+    duplicates = {
+        revision: paths
+        for revision, paths in owners.items()
+        if len(paths) > 1
+    }
+    if not duplicates:
+        return
+
+    details = "; ".join(
+        f"{revision}: {', '.join(str(path) for path in paths)}"
+        for revision, paths in sorted(duplicates.items())
+    )
+    raise ValueError(
+        "Alembic revision IDs are global across enabled modules; duplicate "
+        f"revision declarations found: {details}"
+    )
+
+
+def _literal_migration_revision(path: Path) -> str | None:
+    try:
+        tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
+    except (OSError, SyntaxError, UnicodeError):
+        return None
+    for statement in tree.body:
+        value: ast.expr | None = None
+        if isinstance(statement, ast.Assign) and any(
+            isinstance(target, ast.Name) and target.id == "revision"
+            for target in statement.targets
+        ):
+            value = statement.value
+        elif (
+            isinstance(statement, ast.AnnAssign)
+            and isinstance(statement.target, ast.Name)
+            and statement.target.id == "revision"
+        ):
+            value = statement.value
+        if isinstance(value, ast.Constant) and isinstance(value.value, str):
+            return value.value.strip() or None
+    return None
+
+
 def database_revision(database_url: str | None = None) -> str | None:
     url = database_url or settings.database_url
     engine = create_engine(url)