fix: fixed ecb ids drift between eba dpm releases

src/dpmcore/services/ecb_validations_import.py

@@ -72,6 +72,13 @@ class EcbValidationsImportResult:
 class EcbValidationsImportService:
     """Import ECB validation rules from a CSV export into a DPM database."""
 
+    # Fixed floor for every ID this service assigns, so ECB's IDs stay put
+    # across releases instead of drifting with the native EBA/DPM dataset.
+    _ECB_ID_BASE = 1_030_000_000
+
+    # Required headroom between the native max ID and `_ECB_ID_BASE`.
+    _ECB_ID_BASE_MIN_MARGIN = 1_000_000
+
     def __init__(self, engine: Engine) -> None:
         """Initialise the service with a SQLAlchemy engine."""
         self._engine = engine
@@ -183,10 +190,57 @@ def _get_parent_operation_code(code: str) -> Optional[str]:
         return match.group(1) if match else None
 
     @staticmethod
-    def _next_int_id(session: Session, model: Any, attr_name: str) -> int:
+    def _next_int_id(
+        session: Session,
+        model: Any,
+        attr_name: str,
+        *,
+        floor: Optional[int] = None,
+    ) -> int:
+        """Return the next available ID for ``attr_name`` on ``model``.
+
+        Without ``floor`` this is a plain "max + 1" over the whole table,
+        which drifts release to release as the surrounding EBA/DPM dataset
+        grows. Passing ``floor`` scopes the scan to IDs already inside that
+        reserved block, so the first assigned ID stays the same run after
+        run regardless of how large the rest of the table has become.
+        """
+        column = getattr(model, attr_name)
+        query = session.query(func.max(column))
+        if floor is not None:
+            query = query.filter(column >= floor)
+        current = query.scalar()
+        if current is None:
+            return floor if floor is not None else 1
+        return int(current) + 1
+
+    @staticmethod
+    def _assert_id_floor_margin(
+        session: Session,
+        model: Any,
+        attr_name: str,
+        *,
+        floor: int,
+        min_margin: int,
+    ) -> None:
+        """Fail loudly if native IDs have drifted too close to ``floor``.
+
+        ``floor`` only keeps ECB's IDs stable as long as every *native*
+        (non-ECB) row in the table stays comfortably below it. Only ECB
+        import ever writes IDs `>= floor`, so ``max(id) < floor`` is always
+        the native high-water mark, regardless of whether this run is a
+        fresh import or a re-run over an already-imported database.
+        """
         column = getattr(model, attr_name)
-        current = session.query(func.max(column)).scalar()
-        return int(current or 0) + 1
+        native_max = (
+            session.query(func.max(column)).filter(column < floor).scalar()
+        )
+        if native_max is not None and floor - int(native_max) < min_margin:
+            raise EcbValidationsImportError(
+                f"Native {model.__name__}.{attr_name} max ({native_max})"
+                f" is within {min_margin} of _ECB_ID_BASE ({floor});"
+                " bump _ECB_ID_BASE."
+            )
 
     def _create_operation_concept(
         self,
@@ -383,9 +437,14 @@ def _create_table_references(
         if not table_codes:
             return 0
 
-        next_node_id = self._next_int_id(session, OperationNode, "node_id")
+        next_node_id = self._next_int_id(
+            session, OperationNode, "node_id", floor=self._ECB_ID_BASE
+        )
         next_ref_id = self._next_int_id(
-            session, OperandReference, "operand_reference_id"
+            session,
+            OperandReference,
+            "operand_reference_id",
+            floor=self._ECB_ID_BASE,
         )
 
         node = OperationNode(
@@ -537,14 +596,29 @@ def _import_ecb_validations_df(  # noqa: C901
             default=None,
         )
 
+        for model, attr_name in (
+            (Operation, "operation_id"),
+            (OperationVersion, "operation_vid"),
+            (OperationNode, "node_id"),
+            (OperandReference, "operand_reference_id"),
+        ):
+            self._assert_id_floor_margin(
+                session,
+                model,
+                attr_name,
+                floor=self._ECB_ID_BASE,
+                min_margin=self._ECB_ID_BASE_MIN_MARGIN,
+            )
+
         counters = {
             "operation_id": self._next_int_id(
-                session, Operation, "operation_id"
+                session, Operation, "operation_id", floor=self._ECB_ID_BASE
             ),
             "operation_vid": self._next_int_id(
                 session,
                 OperationVersion,
                 "operation_vid",
+                floor=self._ECB_ID_BASE,
             ),
         }
         operations_created = 0

tests/unit/migration/test_ecb_validations_import.py

@@ -238,6 +238,106 @@ def test_next_int_id_returns_max_plus_one(
         finally:
             session.close()
 
+    def test_next_int_id_with_floor_ignores_rows_below_floor(
+        self, service_with_schema, sqlite_engine_with_schema
+    ):
+        session = sessionmaker(bind=sqlite_engine_with_schema)()
+        try:
+            # Rows below the floor (e.g. another org's growing dataset)
+            # must not influence the floor-scoped block's next ID.
+            session.add(
+                Organisation(org_id=5, name="Org5", acronym="O5", id_prefix=1)
+            )
+            session.flush()
+            result = EcbValidationsImportService._next_int_id(
+                session, Organisation, "org_id", floor=1000
+            )
+            assert result == 1000
+        finally:
+            session.close()
+
+    def test_next_int_id_with_floor_returns_max_plus_one_inside_block(
+        self, service_with_schema, sqlite_engine_with_schema
+    ):
+        session = sessionmaker(bind=sqlite_engine_with_schema)()
+        try:
+            session.add(
+                Organisation(org_id=5, name="Org5", acronym="O5", id_prefix=1)
+            )
+            session.add(
+                Organisation(
+                    org_id=1000, name="Org1000", acronym="O1K", id_prefix=2
+                )
+            )
+            session.flush()
+            result = EcbValidationsImportService._next_int_id(
+                session, Organisation, "org_id", floor=1000
+            )
+            assert result == 1001
+        finally:
+            session.close()
+
+    def test_assert_id_floor_margin_passes_with_sufficient_margin(
+        self, service_with_schema, sqlite_engine_with_schema
+    ):
+        session = sessionmaker(bind=sqlite_engine_with_schema)()
+        try:
+            session.add(
+                Organisation(org_id=5, name="Org5", acronym="O5", id_prefix=1)
+            )
+            session.flush()
+            EcbValidationsImportService._assert_id_floor_margin(
+                session,
+                Organisation,
+                "org_id",
+                floor=1000,
+                min_margin=100,
+            )
+        finally:
+            session.close()
+
+    def test_assert_id_floor_margin_passes_when_no_native_rows(
+        self, service_with_schema, sqlite_engine_with_schema
+    ):
+        session = sessionmaker(bind=sqlite_engine_with_schema)()
+        try:
+            EcbValidationsImportService._assert_id_floor_margin(
+                session,
+                Organisation,
+                "org_id",
+                floor=1000,
+                min_margin=100,
+            )
+        finally:
+            session.close()
+
+    def test_assert_id_floor_margin_raises_when_native_ids_drift_too_close(
+        self, service_with_schema, sqlite_engine_with_schema
+    ):
+        session = sessionmaker(bind=sqlite_engine_with_schema)()
+        try:
+            # Native org_id has drifted within the reserved margin of the
+            # floor: the reservation is no longer a safe anchor.
+            session.add(
+                Organisation(
+                    org_id=950, name="Org950", acronym="O950", id_prefix=1
+                )
+            )
+            session.flush()
+            with pytest.raises(
+                EcbValidationsImportError,
+                match="within 100 of _ECB_ID_BASE",
+            ):
+                EcbValidationsImportService._assert_id_floor_margin(
+                    session,
+                    Organisation,
+                    "org_id",
+                    floor=1000,
+                    min_margin=100,
+                )
+        finally:
+            session.close()
+
     def test_get_or_create_ecb_organisation_creates_new(
         self, service_with_schema, sqlite_engine_with_schema
     ):
@@ -495,6 +595,59 @@ def test_import_csv_duplicate_version_key_skipped(
 
         assert result.operation_versions_created == 1
 
+    def test_operation_ids_stable_despite_growing_dataset(self, tmp_path):
+        """Regression: ECB IDs must not drift as the EBA/DPM dataset grows.
+
+        Before this fix, ECB Operation/OperationVersion IDs were assigned
+        as ``max(id) + 1`` over the *whole* table, so they silently
+        shifted release to release as unrelated EBA validations were
+        added ahead of them. Anchoring to the fixed ``_ECB_ID_BASE`` floor
+        keeps the assigned ID the same regardless of how large the rest
+        of the table has grown.
+        """
+        from dpmcore.orm.base import Base
+        from dpmcore.orm.operations import Operation
+
+        csv_file = tmp_path / "ecb.csv"
+        csv_file.write_text(
+            "vr_code,start_release\nV1,4.0\n", encoding="utf-8"
+        )
+
+        def _assigned_operation_id(dummy_count: int) -> int:
+            engine = create_engine("sqlite:///:memory:")
+            Base.metadata.create_all(engine)
+
+            session = sessionmaker(bind=engine)()
+            for i in range(1, dummy_count + 1):
+                session.add(
+                    Operation(
+                        operation_id=i,
+                        code=f"DUMMY_{i}",
+                        type="validation",
+                        source="hierarchy",
+                    )
+                )
+            session.commit()
+            session.close()
+
+            EcbValidationsImportService(engine).import_csv(str(csv_file))
+
+            session = sessionmaker(bind=engine)()
+            try:
+                operation = (
+                    session.query(Operation)
+                    .filter(Operation.code == "V1")
+                    .one()
+                )
+                return operation.operation_id
+            finally:
+                session.close()
+
+        small_dataset_id = _assigned_operation_id(10)
+        large_dataset_id = _assigned_operation_id(5000)
+
+        assert small_dataset_id == large_dataset_id
+
     def test_extract_table_codes_for_empty_expression_returns_empty_set(
         self, service_with_schema, sqlite_engine_with_schema
     ):