feat(api): only load tenant required for each table

api/services/legacy_model_type_migration.py

@@ -8,9 +8,17 @@ decisions, row updates, row deletes, and structured logging. Only some grouped t
 also add cache cleanup; that includes `provider_models` and
 `provider_model_credentials`. Provider-model-credential groups extend that flow by
 rewriting credential references in provider models and load-balancing configs before
-removing loser credential rows. `load_balancing_model_configs` is the intentional
-exception: it does not group or merge rows, and instead reloads and canonicalizes each
-legacy row independently with row-level cache cleanup.
+removing loser credential rows. `load_balancing_model_configs` stays mostly row-level,
+but it first deduplicates `name="__inherit__"` rows by business key before it
+canonicalizes the remaining legacy rows independently with row-level cache cleanup.
+
+Tenant scheduling has two modes. When callers provide an explicit tenant list, the
+service preserves the original tenant-scoped execution model and runs all selected tables
+for each tenant. When callers omit `tenant_ids`, the service discovers tenant
+ids per table and then runs only that table for the discovered tenants. Most
+tables keep the active `model_types` filter in the discovery query, while
+`load_balancing_model_configs` deliberately uses a whole-table tenant scan so
+that query stays easy to understand.
 """
 
 from __future__ import annotations
@@ -19,8 +27,9 @@ import io
 import json
 import sys
 import threading
+import traceback
 import uuid
-from collections.abc import Iterable, Iterator, Sequence
+from collections.abc import Iterable, Sequence
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from dataclasses import asdict, dataclass
 from datetime import datetime
@@ -35,7 +44,7 @@ from sqlalchemy.sql import select
 from core.helper.model_provider_cache import ProviderCredentialsCache, ProviderCredentialsCacheType
 from graphon.model_runtime.entities.model_entities import ModelType
 from libs.datetime_utils import naive_utc_now
-from models import LoadBalancingModelConfig, ProviderModel, ProviderModelSetting, Tenant, TenantDefaultModel
+from models import LoadBalancingModelConfig, ProviderModel, ProviderModelSetting, TenantDefaultModel
 from models.base import TypeBase
 from models.provider import ProviderModelCredential
 
@@ -95,6 +104,10 @@ def _normalize_log_payload(value: object) -> object:
     return f"<{type(value).__module__}.{type(value).__qualname__}>"
 
 
+def _format_exception_stacktrace(exc: BaseException) -> str:
+    return "".join(traceback.format_exception(type(exc), exc, exc.__traceback__))
+
+
 @dataclass(frozen=True, slots=True)
 class _RowWithRawModelType[T: TypeBase]:
     row: T
@@ -176,6 +189,16 @@ class _ProviderModelSettingBusinessKey(_BusinessKey):
     model_type: ModelType
 
 
+@dataclass(frozen=True, slots=True)
+class _LoadBalancingModelConfigInheritBusinessKey(_BusinessKey):
+    """Business key for `name="__inherit__"` load-balancing configs."""
+
+    tenant_id: str
+    provider_name: str
+    model_name: str
+    model_type: ModelType
+
+
 @dataclass(frozen=True, slots=True)
 class _ProviderModelCredentialBusinessKey(_BusinessKey):
     """Although `ProviderModelCredential` does not have the unique index
@@ -210,6 +233,13 @@ class _ProviderModelSettingGroupPlan:
     loser_rows: list[_RowWithRawModelType[ProviderModelSetting]]
 
 
+@dataclass(frozen=True, slots=True)
+class _LoadBalancingModelConfigInheritGroupPlan:
+    group_row_ids: list[str]
+    winner: _RowWithRawModelType[LoadBalancingModelConfig] | None
+    loser_rows: list[_RowWithRawModelType[LoadBalancingModelConfig]]
+
+
 @dataclass(frozen=True, slots=True)
 class _ProviderModelReferenceRewritePlan:
     row_id: str
@@ -269,6 +299,21 @@ _LOCK_TIMEOUT_FALLBACK_MESSAGES: tuple[str, ...] = (
 _RAW_MODEL_TYPE_COLUMN = "_raw_model_type"
 
 
+def _selected_legacy_values(model_types: Sequence[ModelType]) -> list[str]:
+    legacy_values: list[str] = []
+    for model_type in model_types:
+        legacy_values.extend(_CANONICAL_TO_LEGACY[model_type])
+    return legacy_values
+
+
+def _selected_model_type_values(model_types: Sequence[ModelType]) -> list[str]:
+    model_type_values: list[str] = []
+    for model_type in model_types:
+        model_type_values.append(model_type.value)
+        model_type_values.extend(_CANONICAL_TO_LEGACY[model_type])
+    return list(dict.fromkeys(model_type_values))
+
+
 def _session_factory(engine: sa.Engine) -> Session:
     return Session(bind=engine, expire_on_commit=False)
 
@@ -341,6 +386,12 @@ class LegacyModelTypeMigrationService:
     `provider_model_credentials` is selected, that migration also rewrites references in
     `provider_models` and `load_balancing_model_configs`. Tenant migrations can run in a
     thread pool; JSONL output remains line-safe through a shared synchronized writer.
+
+    If `tenant_ids` is omitted, tenant discovery becomes table-scoped: each selected ORM
+    model loads its own tenant ids, then only that table is dispatched for those tenants.
+    Most tables keep the active model-type filter in discovery, while
+    `load_balancing_model_configs` intentionally uses the whole table so the tenant query
+    stays simple. This still avoids merging tenant ids across unrelated tables.
     """
 
     _engine: sa.Engine
@@ -404,22 +455,51 @@ class LegacyModelTypeMigrationService:
         return tuple(ordered_models)
 
     def migrate(self) -> None:
-        tenant_ids = tuple(self._iter_tenant_ids())
+        output = _ThreadSafeLineWriter(self._output)
+        if self._tenant_ids is not None:
+            self._migrate_explicit_tenants(output)
+            return
+
+        self._migrate_tables_with_discovered_tenants(output)
+
+    def _migrate_explicit_tenants(self, output: io.TextIOBase) -> None:
+        tenant_ids = self._tenant_ids
         if not tenant_ids:
             return
 
-        output = _ThreadSafeLineWriter(self._output)
+        self._run_migrations_for_tenants(tenant_ids, self._orm_models, output)
+
+    def _migrate_tables_with_discovered_tenants(self, output: io.TextIOBase) -> None:
+        for orm_model in self._orm_models:
+            tenant_ids = self._load_tenant_ids_for_model(orm_model)
+            if not tenant_ids:
+                continue
+            self._run_migrations_for_tenants(tenant_ids, (orm_model,), output)
+
+    def _run_migrations_for_tenants(
+        self,
+        tenant_ids: Sequence[str],
+        orm_models: Sequence[ORMModel],
+        output: io.TextIOBase,
+    ) -> None:
         if self._concurrency == 1 or len(tenant_ids) == 1:
             for tenant_id in tenant_ids:
-                self._run_tenant_migration(tenant_id, output)
+                self._run_tenant_migration(tenant_id, orm_models, output)
             return
 
         with ThreadPoolExecutor(max_workers=min(self._concurrency, len(tenant_ids))) as executor:
-            futures = [executor.submit(self._run_tenant_migration, tenant_id, output) for tenant_id in tenant_ids]
+            futures = [
+                executor.submit(self._run_tenant_migration, tenant_id, orm_models, output) for tenant_id in tenant_ids
+            ]
             for future in as_completed(futures):
                 future.result()
 
-    def _run_tenant_migration(self, tenant_id: str, output: io.TextIOBase) -> None:
+    def _run_tenant_migration(
+        self,
+        tenant_id: str,
+        orm_models: Sequence[ORMModel],
+        output: io.TextIOBase,
+    ) -> None:
         """
         Execute one tenant migration with the shared, line-synchronized output stream.
         """
@@ -430,18 +510,88 @@ class LegacyModelTypeMigrationService:
             apply=self._apply,
             output=output,
             model_types=self._model_types,
-            orm_models=self._orm_models,
+            orm_models=orm_models,
         ).run()
 
-    def _iter_tenant_ids(self) -> Iterator[str]:
-        if self._tenant_ids is not None:
-            yield from self._tenant_ids
-            return
+    def _load_tenant_ids_for_model(self, orm_model: ORMModel) -> tuple[str, ...]:
+        """
+        Discover only the tenants that have candidate rows for the current table.
 
+        In automatic tenant mode we keep discovery table-scoped so large shared tenant
+        populations do not force empty work for unrelated tables. Most table queries
+        still apply the active `model_types` filter before scheduling migrations, while
+        `load_balancing_model_configs` intentionally trades a wider tenant set for a
+        simpler discovery query.
+        """
+
+        legacy_model_type_values = _selected_legacy_values(self._model_types)
         with _session_factory(self._engine) as session:
-            tenant_ids = session.execute(select(Tenant.id).order_by(Tenant.id.asc())).scalars().all()
+            if orm_model is ProviderModel:
+                tenant_ids = (
+                    session.execute(
+                        select(ProviderModel.tenant_id)
+                        .where(sa.type_coerce(ProviderModel.model_type, sa.String()).in_(legacy_model_type_values))
+                        .distinct()
+                        .order_by(ProviderModel.tenant_id.asc())
+                    )
+                    .scalars()
+                    .all()
+                )
+            elif orm_model is TenantDefaultModel:
+                tenant_ids = (
+                    session.execute(
+                        select(TenantDefaultModel.tenant_id)
+                        .where(sa.type_coerce(TenantDefaultModel.model_type, sa.String()).in_(legacy_model_type_values))
+                        .distinct()
+                        .order_by(TenantDefaultModel.tenant_id.asc())
+                    )
+                    .scalars()
+                    .all()
+                )
+            elif orm_model is ProviderModelSetting:
+                tenant_ids = (
+                    session.execute(
+                        select(ProviderModelSetting.tenant_id)
+                        .where(
+                            sa.type_coerce(ProviderModelSetting.model_type, sa.String()).in_(legacy_model_type_values)
+                        )
+                        .distinct()
+                        .order_by(ProviderModelSetting.tenant_id.asc())
+                    )
+                    .scalars()
+                    .all()
+                )
+            elif orm_model is LoadBalancingModelConfig:
+                # Deliberately discover tenants from the whole table so the query stays
+                # easier to understand than the legacy/canonical mixed-row filter.
+                tenant_ids = (
+                    session.execute(
+                        select(LoadBalancingModelConfig.tenant_id)
+                        .distinct()
+                        .order_by(LoadBalancingModelConfig.tenant_id.asc())
+                    )
+                    .scalars()
+                    .all()
+                )
+            elif orm_model is ProviderModelCredential:
+                tenant_ids = (
+                    session.execute(
+                        select(ProviderModelCredential.tenant_id)
+                        .where(
+                            sa.type_coerce(ProviderModelCredential.model_type, sa.String()).in_(
+                                legacy_model_type_values
+                            )
+                        )
+                        .distinct()
+                        .order_by(ProviderModelCredential.tenant_id.asc())
+                    )
+                    .scalars()
+                    .all()
+                )
+            else:
+                raise ValueError(f"unsupported orm model: {orm_model}")
 
-        yield from tenant_ids
+        return tuple(tenant_ids)
 
 
 class Migration:
@@ -510,14 +660,28 @@ class Migration:
         )
 
     def _selected_legacy_values(self) -> list[str]:
-        legacy_values: list[str] = []
-        for model_type in self._model_types:
-            legacy_values.extend(_CANONICAL_TO_LEGACY[model_type])
-        return legacy_values
+        return _selected_legacy_values(self._model_types)
+
+    def _selected_model_type_values(self) -> list[str]:
+        return _selected_model_type_values(self._model_types)
 
     def _allowed_values_for_canonical_model_type(self, canonical_model_type: ModelType) -> tuple[str, ...]:
         return (*_CANONICAL_TO_LEGACY[canonical_model_type], canonical_model_type.value)
 
+    def _normalize_selected_model_type(self, raw_model_type: str) -> ModelType | None:
+        canonical_model_type = _LEGACY_TO_CANONICAL.get(raw_model_type)
+        if canonical_model_type is not None:
+            return canonical_model_type
+
+        try:
+            parsed_model_type = ModelType(raw_model_type)
+        except ValueError:
+            return None
+
+        if parsed_model_type not in self._model_types:
+            return None
+        return parsed_model_type
+
     def _has_legacy_rows[T: TypeBase](self, rows: Sequence[_RowWithRawModelType[T]]) -> bool:
         return any(row.raw_model_type in _LEGACY_TO_CANONICAL for row in rows)
 
@@ -1195,9 +1359,11 @@ class Migration:
         """
         Migrate load-balancing configs row by row.
 
-        This table only needs model_type canonicalization. Unlike the grouped tables, it
-        must not merge rows by business key; each legacy candidate is reloaded and updated
-        independently so the migration remains a pure per-row rewrite plus cache cleanup.
+        This table first deduplicates `name="__inherit__"` rows per normalized
+        `(tenant_id, provider_name, model_name, model_type)` business key, then
+        canonicalizes the remaining legacy rows independently. The pre-pass must run
+        first so a legacy/canonical `__inherit__` pair keeps only the newest row before
+        the row-level canonicalization would collapse them onto the same canonical key.
         """
         self._log_event(
             "table_started",
@@ -1209,6 +1375,7 @@ class Migration:
             },
         )
 
+        processed_inherit_groups = self._deduplicate_inherit_load_balancing_model_configs()
         processed_rows = 0
         last_id: str | None = None
 
@@ -1229,10 +1396,217 @@ class Migration:
                 "tenant_id": self._tenant_id,
                 "apply": self._apply,
                 "table_name": LoadBalancingModelConfig.__tablename__,
+                "processed_inherit_groups": processed_inherit_groups,
                 "processed_rows": processed_rows,
             },
         )
 
+    def _deduplicate_inherit_load_balancing_model_configs(self) -> int:
+        seen_business_keys: dict[_LoadBalancingModelConfigInheritBusinessKey, list[str]] = {}
+        processed_groups = 0
+        last_id: str | None = None
+
+        while True:
+            candidates = self._load_load_balancing_inherit_candidates(last_id)
+            if not candidates:
+                break
+
+            for candidate in candidates:
+                last_id = str(candidate.row.id)
+                business_key = _LoadBalancingModelConfigInheritBusinessKey(
+                    tenant_id=candidate.row.tenant_id,
+                    provider_name=candidate.row.provider_name,
+                    model_name=candidate.row.model_name,
+                    model_type=candidate.canonical_model_type,
+                )
+                if business_key in seen_business_keys:
+                    continue
+
+                seen_business_keys[business_key] = self._process_load_balancing_inherit_group(candidate, business_key)
+                processed_groups += 1
+
+        return processed_groups
+
+    def _load_load_balancing_inherit_candidates(
+        self, last_id: str | None
+    ) -> list[_RowWithRawModelType[LoadBalancingModelConfig]]:
+        raw_model_type = sa.type_coerce(LoadBalancingModelConfig.model_type, sa.String()).label(_RAW_MODEL_TYPE_COLUMN)
+        with _session_factory(self._engine) as session:
+            stmt = (
+                select(LoadBalancingModelConfig, raw_model_type)
+                .where(
+                    LoadBalancingModelConfig.tenant_id == self._tenant_id,
+                    LoadBalancingModelConfig.name == "__inherit__",
+                    sa.type_coerce(LoadBalancingModelConfig.model_type, sa.String()).in_(
+                        self._selected_model_type_values()
+                    ),
+                )
+                .order_by(LoadBalancingModelConfig.id.asc())
+                .limit(self._batch_size)
+            )
+            if last_id is not None:
+                stmt = stmt.where(LoadBalancingModelConfig.id > last_id)
+            rows = session.execute(stmt).all()
+
+        wrapped_rows: list[_RowWithRawModelType[LoadBalancingModelConfig]] = []
+        for load_balancing_model_config, raw_value in rows:
+            raw_model_type_value = str(raw_value)
+            canonical_model_type = self._normalize_selected_model_type(raw_model_type_value)
+            if canonical_model_type is None:
+                self._log_event(
+                    event="invalid_model_type",
+                    message=f"invalid model type: {raw_value}",
+                    attrs={
+                        "id": load_balancing_model_config.id,
+                        "table_name": load_balancing_model_config.__tablename__,
+                    },
+                )
+                continue
+
+            wrapped_rows.append(
+                _RowWithRawModelType(
+                    row=load_balancing_model_config,
+                    raw_model_type=raw_model_type_value,
+                    canonical_model_type=canonical_model_type,
+                )
+            )
+        return wrapped_rows
+
+    def _load_load_balancing_inherit_group(
+        self,
+        session: Session,
+        candidate: _RowWithRawModelType[LoadBalancingModelConfig],
+        *,
+        lock_rows: bool,
+    ) -> list[_RowWithRawModelType[LoadBalancingModelConfig]]:
+        raw_model_type = sa.type_coerce(LoadBalancingModelConfig.model_type, sa.String()).label(_RAW_MODEL_TYPE_COLUMN)
+        stmt = (
+            select(LoadBalancingModelConfig, raw_model_type)
+            .where(
+                LoadBalancingModelConfig.tenant_id == candidate.row.tenant_id,
+                LoadBalancingModelConfig.provider_name == candidate.row.provider_name,
+                LoadBalancingModelConfig.model_name == candidate.row.model_name,
+                LoadBalancingModelConfig.name == "__inherit__",
+                sa.type_coerce(LoadBalancingModelConfig.model_type, sa.String()).in_(
+                    self._allowed_values_for_canonical_model_type(candidate.canonical_model_type)
+                ),
+            )
+            .order_by(LoadBalancingModelConfig.id.asc())
+        )
+        if lock_rows:
+            stmt = stmt.with_for_update()
+
+        rows = session.execute(stmt).all()
+        wrapped_rows: list[_RowWithRawModelType[LoadBalancingModelConfig]] = []
+        for load_balancing_model_config, raw_value in rows:
+            raw_model_type_value = str(raw_value)
+            canonical_model_type = self._normalize_selected_model_type(raw_model_type_value)
+            if canonical_model_type is None:
+                continue
+            wrapped_rows.append(
+                _RowWithRawModelType(
+                    row=load_balancing_model_config,
+                    raw_model_type=raw_model_type_value,
+                    canonical_model_type=canonical_model_type,
+                )
+            )
+        return wrapped_rows
+
+    def _build_load_balancing_inherit_group_plan(
+        self,
+        session: Session,
+        candidate: _RowWithRawModelType[LoadBalancingModelConfig],
+        *,
+        lock_rows: bool,
+    ) -> _LoadBalancingModelConfigInheritGroupPlan:
+        rows = self._load_load_balancing_inherit_group(session, candidate, lock_rows=lock_rows)
+        group_row_ids = [str(row.row.id) for row in rows]
+        if len(rows) <= 1:
+            return _LoadBalancingModelConfigInheritGroupPlan(group_row_ids=group_row_ids, winner=None, loser_rows=[])
+
+        winner = self._select_winner(rows)
+        return _LoadBalancingModelConfigInheritGroupPlan(
+            group_row_ids=group_row_ids,
+            winner=winner,
+            loser_rows=[row for row in rows if row.row.id != winner.row.id],
+        )
+
+    def _emit_load_balancing_inherit_group_plan(
+        self,
+        plan: _LoadBalancingModelConfigInheritGroupPlan,
+        *,
+        session: Session,
+        tx_id: str,
+        business_key: _LoadBalancingModelConfigInheritBusinessKey,
+    ) -> None:
+        if plan.winner is None:
+            return
+
+        cache_plans: list[_CacheDeletePlan] = []
+        for loser in plan.loser_rows:
+            loser_row_id = str(loser.row.id)
+            if self._apply:
+                session.execute(sa.delete(LoadBalancingModelConfig).where(LoadBalancingModelConfig.id == loser_row_id))
+            self._log_row_deleted(
+                LoadBalancingModelConfig.__tablename__,
+                loser,
+                tx_id=tx_id,
+                business_key=business_key,
+                related_winner_id=str(plan.winner.row.id),
+            )
+            cache_plans.append(
+                _CacheDeletePlan(
+                    tenant_id=self._tenant_id,
+                    identity_id=loser_row_id,
+                    cache_type=ProviderCredentialsCacheType.LOAD_BALANCING_MODEL,
+                    table_name=LoadBalancingModelConfig.__tablename__,
+                    row_id=loser_row_id,
+                    tx_id=tx_id,
+                    business_key=business_key,
+                )
+            )
+
+        self._log_cache_plans(cache_plans, apply=self._apply)
+        self._log_group_processed(
+            LoadBalancingModelConfig.__tablename__,
+            business_key,
+            plan.group_row_ids,
+            tx_id=tx_id,
+        )
+
+    def _process_load_balancing_inherit_group(
+        self,
+        candidate: _RowWithRawModelType[LoadBalancingModelConfig],
+        business_key: _LoadBalancingModelConfigInheritBusinessKey,
+    ) -> list[str]:
+        tx_id = self._new_tx_id()
+        group_row_ids = [str(candidate.row.id)]
+
+        try:
+            with _session_factory(self._engine) as session, session.begin():
+                self._configure_lock_timeout(session)
+                plan = self._build_load_balancing_inherit_group_plan(session, candidate, lock_rows=True)
+                group_row_ids = plan.group_row_ids or group_row_ids
+                self._emit_load_balancing_inherit_group_plan(
+                    plan,
+                    session=session,
+                    tx_id=tx_id,
+                    business_key=business_key,
+                )
+        except OperationalError as exc:
+            if self._is_lock_timeout_error(exc):
+                self._log_lock_timeout(
+                    LoadBalancingModelConfig.__tablename__,
+                    str(candidate.row.id),
+                    tx_id,
+                    business_key,
+                    exc,
+                )
+                return group_row_ids
+            raise
+
+        return group_row_ids
+
     def _load_load_balancing_model_config_candidates(
         self, last_id: str | None
     ) -> list[_RowWithRawModelType[LoadBalancingModelConfig]]:
@@ -1336,10 +1710,11 @@ class Migration:
             ).delete()
             self._log_event("cache_deleted", "Deleted related cache entry.", attrs)
         except Exception as exc:
-            self._log_event(
+            self._log_exception_event(
                 "cache_delete_failed",
                 "Failed to delete related cache entry.",
-                {**attrs, "error": str(exc)},
+                attrs,
+                exc,
             )
 
     def _process_load_balancing_model_config_row(
@@ -1902,11 +2277,15 @@ class Migration:
             "table_name": table_name,
             "id": row_id,
             "tx_id": tx_id,
-            "error": str(exc),
         }
         if business_key is not None:
             attrs["business_key"] = self._business_key_to_dict(business_key)
-        self._log_event("lock_timeout_skipped", "Skipped transaction because row lock timed out.", attrs)
+        self._log_exception_event(
+            "lock_timeout_skipped",
+            "Skipped transaction because row lock timed out.",
+            attrs,
+            exc,
+        )
 
     def _business_key_to_dict(self, business_key: _BusinessKey) -> dict[str, object]:
         return cast(dict[str, object], asdict(business_key))
@@ -2012,7 +2391,7 @@ class Migration:
                         },
                     )
                 except Exception as exc:
-                    self._log_event(
+                    self._log_exception_event(
                         "cache_delete_failed",
                         "Failed to delete related cache entry.",
                         {
@@ -2023,8 +2402,8 @@ class Migration:
                             "cache_type": cache_plan.cache_type.value,
                             "tx_id": cache_plan.tx_id,
                             "business_key": self._business_key_to_dict(cache_plan.business_key),
-                            "error": str(exc),
                         },
+                        exc,
                     )
             else:
                 self._log_event(
@@ -2041,6 +2420,23 @@ class Migration:
                     },
                 )
 
+    def _log_exception_event(
+        self,
+        event: str,
+        message: str,
+        attrs: dict[str, object],
+        exc: BaseException,
+    ) -> None:
+        self._log_event(
+            event,
+            message,
+            {
+                **attrs,
+                "error": str(exc),
+                "stacktrace": _format_exception_stacktrace(exc),
+            },
+        )
+
     def _log_event(self, event: str, message: str, attrs: dict[str, object]) -> None:
         record = {
             "event": event,

api/tests/test_containers_integration_tests/commands/test_legacy_model_type_migration.py

@@ -2,7 +2,9 @@ from __future__ import annotations
 
 import importlib
 import io
+import json
 from collections.abc import Generator
+from datetime import datetime, timedelta
 
 import pytest
 import sqlalchemy as sa
@@ -15,6 +17,100 @@ from tests.helpers.legacy_model_type_migration import (
 )
 
 
+def _parse_json_lines(output: io.StringIO) -> list[dict[str, object]]:
+    return [json.loads(line) for line in output.getvalue().splitlines() if line.strip()]
+
+
+def _json_key(value: object) -> str:
+    return json.dumps(value, sort_keys=True)
+
+
+def _lb_processing_signatures(lines: list[dict[str, object]]) -> set[tuple[object, ...]]:
+    signatures: set[tuple[object, ...]] = set()
+    for line in lines:
+        attrs = line.get("attrs")
+        if not isinstance(attrs, dict):
+            continue
+        if attrs.get("table_name") != "load_balancing_model_configs":
+            continue
+        event = line.get("event")
+        if event == "row_updated":
+            signatures.add(
+                (
+                    event,
+                    attrs.get("id"),
+                    _json_key(attrs.get("old_values")),
+                    _json_key(attrs.get("new_values")),
+                )
+            )
+        elif event == "row_deleted":
+            signatures.add(
+                (
+                    event,
+                    attrs.get("id"),
+                    attrs.get("merge_winner_id"),
+                )
+            )
+        elif event == "group_processed":
+            signatures.add(
+                (
+                    event,
+                    attrs.get("table_name"),
+                    _json_key(attrs.get("business_key")),
+                    tuple(attrs.get("group_row_ids", [])),
+                )
+            )
+    return signatures
+
+
+def _insert_load_balancing_model_config(
+    engine: sa.Engine,
+    *,
+    row_id: str,
+    tenant_id: str,
+    provider_name: str,
+    model_name: str,
+    model_type: str,
+    name: str,
+    encrypted_config: str,
+    credential_id: str,
+    enabled: bool,
+    created_at: datetime,
+    updated_at: datetime,
+) -> None:
+    with engine.begin() as conn:
+        conn.execute(
+            sa.text(
+                """
+                INSERT INTO load_balancing_model_configs
+                    (
+                        id, tenant_id, provider_name, model_name, model_type, name,
+                        encrypted_config, credential_id, credential_source_type, enabled, created_at, updated_at
+                    )
+                VALUES
+                    (
+                        :id, :tenant_id, :provider_name, :model_name, :model_type, :name,
+                        :encrypted_config, :credential_id, :credential_source_type, :enabled, :created_at, :updated_at
+                    )
+                """
+            ),
+            {
+                "id": row_id,
+                "tenant_id": tenant_id,
+                "provider_name": provider_name,
+                "model_name": model_name,
+                "model_type": model_type,
+                "name": name,
+                "encrypted_config": encrypted_config,
+                "credential_id": credential_id,
+                "credential_source_type": "custom_model",
+                "enabled": enabled,
+                "created_at": created_at,
+                "updated_at": updated_at,
+            },
+        )
+
+
 @pytest.fixture(scope="session")
 def migration_module():
     try:
@@ -125,3 +221,188 @@ def test_legacy_model_type_migration_end_to_end_across_supported_backends(
         for table_name in first_apply_state
     }
     assert second_apply_state == first_apply_state
+
+
+def test_load_balancing_inherit_deduplication_is_applied_consistently_across_supported_backends(
+    migration_module,
+    container_engine: tuple[str, sa.Engine],
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    _, engine = container_engine
+    helper_module = importlib.import_module("tests.helpers.legacy_model_type_migration")
+    helper_module.drop_minimal_legacy_model_type_schema(engine)
+    fixture = seed_legacy_model_type_dirty_data(engine)
+
+    tenant_id = fixture.primary.tenant_id
+    older_inherit_row_id = "00000000-0000-0000-0000-00000000ee01"
+    newer_inherit_row_id = "00000000-0000-0000-0000-00000000ee02"
+    canonical_non_inherit_row_id = "00000000-0000-0000-0000-00000000ee03"
+    created_at = datetime(2025, 1, 1, 8, 0, 0)
+
+    _insert_load_balancing_model_config(
+        engine,
+        row_id=older_inherit_row_id,
+        tenant_id=tenant_id,
+        provider_name="openai",
+        model_name="gpt-4o-mini",
+        model_type="llm",
+        name="__inherit__",
+        encrypted_config='{"api_key":"older-inherit"}',
+        credential_id=fixture.primary.winner_credential_id,
+        enabled=True,
+        created_at=created_at,
+        updated_at=created_at + timedelta(minutes=15),
+    )
+    _insert_load_balancing_model_config(
+        engine,
+        row_id=newer_inherit_row_id,
+        tenant_id=tenant_id,
+        provider_name="openai",
+        model_name="gpt-4o-mini",
+        model_type="text-generation",
+        name="__inherit__",
+        encrypted_config='{"api_key":"newer-inherit"}',
+        credential_id=fixture.primary.distinct_credential_id,
+        enabled=True,
+        created_at=created_at,
+        updated_at=created_at + timedelta(minutes=30),
+    )
+    _insert_load_balancing_model_config(
+        engine,
+        row_id=canonical_non_inherit_row_id,
+        tenant_id=tenant_id,
+        provider_name="openai",
+        model_name="gpt-4o-mini",
+        model_type="llm",
+        name=f"{tenant_id}-second-shared",
+        encrypted_config='{"api_key":"non-inherit-canonical"}',
+        credential_id=fixture.primary.distinct_credential_id,
+        enabled=True,
+        created_at=created_at,
+        updated_at=created_at + timedelta(minutes=45),
+    )
+
+    before_dry_run = fetch_table_rows(engine, "load_balancing_model_configs", tenant_id=tenant_id)
+    deleted_cache_keys: list[str] = []
+
+    def _record_delete(self) -> None:
+        deleted_cache_keys.append(self.cache_key)
+
+    monkeypatch.setattr(migration_module.ProviderCredentialsCache, "delete", _record_delete)
+
+    dry_run_output = io.StringIO()
+    migration_module.LegacyModelTypeMigrationService(
+        engine=engine,
+        apply=False,
+        output=dry_run_output,
+        tables=("load_balancing_model_configs",),
+        model_types=(migration_module.ModelType.LLM,),
+        tenant_ids=(tenant_id,),
+    ).migrate()
+
+    after_dry_run = fetch_table_rows(engine, "load_balancing_model_configs", tenant_id=tenant_id)
+    dry_run_lines = _parse_json_lines(dry_run_output)
+    dry_run_cache_events = [line["event"] for line in dry_run_lines if str(line.get("event")).startswith("cache_")]
+    dry_run_row_updates = {
+        str(attrs["id"])
+        for line in dry_run_lines
+        if line.get("event") == "row_updated"
+        and isinstance((attrs := line.get("attrs")), dict)
+        and attrs.get("table_name") == "load_balancing_model_configs"
+    }
+    dry_run_row_deletes = {
+        str(attrs["id"])
+        for line in dry_run_lines
+        if line.get("event") == "row_deleted"
+        and isinstance((attrs := line.get("attrs")), dict)
+        and attrs.get("table_name") == "load_balancing_model_configs"
+    }
+    dry_run_group_processed = [
+        attrs
+        for line in dry_run_lines
+        if line.get("event") == "group_processed"
+        and isinstance((attrs := line.get("attrs")), dict)
+        and attrs.get("table_name") == "load_balancing_model_configs"
+    ]
+
+    assert after_dry_run == before_dry_run
+    assert deleted_cache_keys == []
+    assert dry_run_row_deletes == {older_inherit_row_id}
+    assert dry_run_row_updates == {
+        fixture.primary.load_balancing_config_id,
+        newer_inherit_row_id,
+    }
+    assert canonical_non_inherit_row_id not in dry_run_row_updates
+    assert "cache_delete_planned" in dry_run_cache_events
+    assert "cache_deleted" not in dry_run_cache_events
+    assert len(dry_run_group_processed) == 1
+    assert dry_run_group_processed[0]["table_name"] == "load_balancing_model_configs"
+    assert dry_run_group_processed[0]["business_key"] == {
+        "tenant_id": tenant_id,
+        "provider_name": "openai",
+        "model_name": "gpt-4o-mini",
+        "model_type": "llm",
+    }
+    assert set(dry_run_group_processed[0]["group_row_ids"]) == {
+        older_inherit_row_id,
+        newer_inherit_row_id,
+    }
+
+    apply_output = io.StringIO()
+    migration_module.LegacyModelTypeMigrationService(
+        engine=engine,
+        apply=True,
+        output=apply_output,
+        tables=("load_balancing_model_configs",),
+        model_types=(migration_module.ModelType.LLM,),
+        tenant_ids=(tenant_id,),
+    ).migrate()
+
+    apply_lines = _parse_json_lines(apply_output)
+    apply_cache_events = [line["event"] for line in apply_lines if str(line.get("event")).startswith("cache_")]
+    apply_group_processed = [
+        attrs
+        for line in apply_lines
+        if line.get("event") == "group_processed"
+        and isinstance((attrs := line.get("attrs")), dict)
+        and attrs.get("table_name") == "load_balancing_model_configs"
+    ]
+    assert _lb_processing_signatures(apply_lines) == _lb_processing_signatures(dry_run_lines)
+    assert "cache_deleted" in apply_cache_events
+    assert deleted_cache_keys
+    assert len(apply_group_processed) == len(dry_run_group_processed)
+    assert [
+        (
+            attrs["table_name"],
+            _json_key(attrs["business_key"]),
+            tuple(attrs["group_row_ids"]),
+        )
+        for attrs in apply_group_processed
+    ] == [
+        (
+            attrs["table_name"],
+            _json_key(attrs["business_key"]),
+            tuple(attrs["group_row_ids"]),
+        )
+        for attrs in dry_run_group_processed
+    ]
+
+    lb_rows = fetch_table_rows(engine, "load_balancing_model_configs", tenant_id=tenant_id)
+    surviving_inherit_rows = [row for row in lb_rows if row["name"] == "__inherit__"]
+    surviving_non_inherit_rows = [row for row in lb_rows if row["name"] != "__inherit__"]
+
+    assert {str(row["id"]) for row in surviving_inherit_rows} == {newer_inherit_row_id}
+    assert surviving_inherit_rows[0]["model_type"] == "llm"
+    assert surviving_inherit_rows[0]["credential_id"] == fixture.primary.distinct_credential_id
+
+    assert {
+        str(row["id"])
+        for row in surviving_non_inherit_rows
+        if str(row["id"]) in {fixture.primary.load_balancing_config_id, canonical_non_inherit_row_id}
+    } == {fixture.primary.load_balancing_config_id, canonical_non_inherit_row_id}
+    assert all(
+        row["model_type"] == "llm"
+        for row in surviving_non_inherit_rows
+        if str(row["id"]) in {fixture.primary.load_balancing_config_id, canonical_non_inherit_row_id}
+    )
+    assert count_rows(engine, "load_balancing_model_configs", tenant_id=tenant_id) == len(before_dry_run) - 1

api/tests/unit_tests/commands/test_legacy_model_type_migration.py

@@ -17,6 +17,7 @@ from click.testing import CliRunner
 from sqlalchemy.exc import OperationalError
 
 from graphon.model_runtime.entities.model_entities import ModelType
+from models.account import Tenant
 from models.enums import CredentialSourceType
 from models.provider import ProviderModel
 from tests.helpers.legacy_model_type_migration import (
@@ -24,6 +25,7 @@ from tests.helpers.legacy_model_type_migration import (
     LEGACY_TO_CANONICAL,
     assert_tenant_rows_use_only_canonical_model_types,
     count_rows,
+    create_minimal_legacy_model_type_schema,
     fetch_table_rows,
     seed_legacy_model_type_dirty_data,
     snapshot_legacy_model_type_state,
@@ -117,6 +119,28 @@ def _collect_processing_signatures(lines: list[dict[str, object]]) -> set[tuple[
     return signatures
 
 
+def _cache_event_row_ids(
+    lines: list[dict[str, object]],
+    *,
+    table_name: str,
+    row_ids: set[str],
+    event_name: str,
+) -> set[str]:
+    matching_row_ids: set[str] = set()
+    for line in lines:
+        if line.get("event") != event_name:
+            continue
+        attrs = line.get("attrs")
+        if not isinstance(attrs, dict):
+            continue
+        if attrs.get("table_name") != table_name:
+            continue
+        row_id = str(attrs.get("id"))
+        if row_id in row_ids:
+            matching_row_ids.add(row_id)
+    return matching_row_ids
+
+
 def _patch_batch_size(
     monkeypatch: pytest.MonkeyPatch,
     migration_module,
@@ -174,6 +198,18 @@ def _insert_provider_model(
         )
 
 
+def _insert_tenant(engine: sa.Engine, *, tenant_id: str) -> None:
+    with engine.begin() as conn:
+        conn.execute(
+            Tenant.__table__.insert().values(
+                id=tenant_id,
+                name=f"Tenant {tenant_id}",
+                plan="basic",
+                status="normal",
+            )
+        )
+
+
 def _insert_tenant_default_model(
     engine: sa.Engine,
     *,
@@ -487,7 +523,7 @@ def test_service_migrate_batches_by_tenant_respects_selected_tables_without_reve
     migration_module,
     sqlite_engine: sa.Engine,
 ) -> None:
-    seen_runs: list[dict[str, object]] = []
+    seen_runs: list[tuple[str, tuple[str, ...], tuple[ModelType, ...]]] = []
 
     class FakeMigration:
         def __init__(
@@ -500,18 +536,12 @@ def test_service_migrate_batches_by_tenant_respects_selected_tables_without_reve
             model_types: tuple[ModelType, ...],
             orm_models: tuple[type[object], ...],
         ) -> None:
-            seen_runs.append(
-                {
-                    "tenant_id": tenant_id,
-                    "engine": engine,
-                    "apply": apply,
-                    "model_types": model_types,
-                    "table_names": tuple(model.__table__.name for model in orm_models),
-                }
-            )
+            assert engine is sqlite_engine
+            assert apply is False
+            seen_runs.append((tenant_id, tuple(model.__table__.name for model in orm_models), model_types))
 
         def run(self) -> None:
-            seen_runs.append({"run": True})
+            return None
 
     monkeypatch = pytest.MonkeyPatch()
     try:
@@ -520,7 +550,7 @@ def test_service_migrate_batches_by_tenant_respects_selected_tables_without_reve
             engine=sqlite_engine,
             apply=False,
             concurrency=1,
-            tables=("provider_models",),
+            tables=("provider_models", "tenant_default_models"),
             model_types=(ModelType.LLM,),
             tenant_ids=("tenant-alpha", "tenant-beta"),
         )
@@ -529,11 +559,267 @@ def test_service_migrate_batches_by_tenant_respects_selected_tables_without_reve
     finally:
         monkeypatch.undo()
 
-    init_calls = [call for call in seen_runs if "tenant_id" in call]
-    assert [call["tenant_id"] for call in init_calls] == ["tenant-alpha", "tenant-beta"]
-    for call in init_calls:
-        assert tuple(cast(tuple[str, ...], call["table_names"])) == ("provider_models",)
-        assert call["model_types"] == (ModelType.LLM,)
+    assert seen_runs == [
+        ("tenant-alpha", ("provider_models", "tenant_default_models"), (ModelType.LLM,)),
+        ("tenant-beta", ("provider_models", "tenant_default_models"), (ModelType.LLM,)),
+    ]
+
+
+def test_service_migrate_without_tenant_ids_discovers_tenants_per_selected_table_without_querying_tenants(
+    migration_module,
+    sqlite_engine: sa.Engine,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    create_minimal_legacy_model_type_schema(sqlite_engine)
+    provider_tenant_id = "00000000-0000-0000-0000-000000000111"
+    default_tenant_id = "00000000-0000-0000-0000-000000000222"
+    empty_tenant_id = "00000000-0000-0000-0000-000000000333"
+    for tenant_id in (provider_tenant_id, default_tenant_id, empty_tenant_id):
+        _insert_tenant(sqlite_engine, tenant_id=tenant_id)
+
+    created_at = datetime(2025, 1, 1, 12, 0, 0)
+    updated_at = created_at + timedelta(minutes=1)
+    _insert_provider_model(
+        sqlite_engine,
+        row_id="10000000-0000-0000-0000-000000000111",
+        tenant_id=provider_tenant_id,
+        provider_name="openai",
+        model_name="gpt-4o-mini",
+        model_type="text-generation",
+        credential_id=None,
+        created_at=created_at,
+        updated_at=updated_at,
+    )
+    _insert_tenant_default_model(
+        sqlite_engine,
+        row_id="20000000-0000-0000-0000-000000000222",
+        tenant_id=default_tenant_id,
+        provider_name="openai",
+        model_name="gpt-4o-mini",
+        model_type="text-generation",
+        created_at=created_at,
+        updated_at=updated_at,
+    )
+
+    seen_runs: list[tuple[str, tuple[str, ...], tuple[ModelType, ...]]] = []
+    executed_sql: list[str] = []
+
+    class FakeMigration:
+        def __init__(
+            self,
+            *,
+            tenant_id: str,
+            engine: sa.Engine,
+            apply: bool,
+            output: io.TextIOBase,
+            model_types: tuple[ModelType, ...],
+            orm_models: tuple[type[object], ...],
+        ) -> None:
+            assert engine is sqlite_engine
+            assert apply is False
+            seen_runs.append((tenant_id, tuple(model.__table__.name for model in orm_models), model_types))
+
+        def run(self) -> None:
+            return None
+
+    def _record_sql(
+        conn: sa.engine.Connection,
+        cursor: object,
+        statement: str,
+        parameters: object,
+        context: object,
+        executemany: bool,
+    ) -> None:
+        del conn, cursor, parameters, context, executemany
+        executed_sql.append(statement)
+
+    sa.event.listen(sqlite_engine, "before_cursor_execute", _record_sql)
+    try:
+        monkeypatch.setattr(migration_module, "Migration", FakeMigration)
+        service = migration_module.LegacyModelTypeMigrationService(
+            engine=sqlite_engine,
+            apply=False,
+            tables=("provider_models", "tenant_default_models"),
+            model_types=(ModelType.LLM,),
+        )
+
+        service.migrate()
+    finally:
+        sa.event.remove(sqlite_engine, "before_cursor_execute", _record_sql)
+
+    assert seen_runs == [
+        (provider_tenant_id, ("provider_models",), (ModelType.LLM,)),
+        (default_tenant_id, ("tenant_default_models",), (ModelType.LLM,)),
+    ]
+    normalized_statements = [" ".join(statement.lower().split()) for statement in executed_sql]
+    discovery_statements = [statement for statement in normalized_statements if statement.startswith("select")]
+    table_names = ("provider_models", "tenant_default_models")
+    table_discovery_statements = [
+        statement
+        for statement in discovery_statements
+        if any(f" from {table_name} " in f" {statement} " for table_name in table_names)
+    ]
+
+    assert [statement for statement in discovery_statements if " from tenants " in f" {statement} "] == []
+    assert [statement for statement in discovery_statements if " union " in f" {statement} "] == []
+    assert [
+        next(table_name for table_name in table_names if f" from {table_name} " in f" {statement} ")
+        for statement in table_discovery_statements
+    ] == list(table_names)
+
+
+def test_service_migrate_without_tenant_ids_filters_provider_model_tenants_by_selected_model_types(
+    migration_module,
+    sqlite_engine: sa.Engine,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    create_minimal_legacy_model_type_schema(sqlite_engine)
+    llm_tenant_id = "00000000-0000-0000-0000-000000000411"
+    embedding_tenant_id = "00000000-0000-0000-0000-000000000422"
+    empty_tenant_id = "00000000-0000-0000-0000-000000000433"
+    for tenant_id in (llm_tenant_id, embedding_tenant_id, empty_tenant_id):
+        _insert_tenant(sqlite_engine, tenant_id=tenant_id)
+
+    created_at = datetime(2025, 1, 2, 12, 0, 0)
+    updated_at = created_at + timedelta(minutes=1)
+    _insert_provider_model(
+        sqlite_engine,
+        row_id="30000000-0000-0000-0000-000000000411",
+        tenant_id=llm_tenant_id,
+        provider_name="openai",
+        model_name="gpt-4o-mini",
+        model_type="text-generation",
+        credential_id=None,
+        created_at=created_at,
+        updated_at=updated_at,
+    )
+    _insert_provider_model(
+        sqlite_engine,
+        row_id="30000000-0000-0000-0000-000000000422",
+        tenant_id=embedding_tenant_id,
+        provider_name="openai",
+        model_name="text-embedding-3-large",
+        model_type="embeddings",
+        credential_id=None,
+        created_at=created_at,
+        updated_at=updated_at,
+    )
+
+    seen_runs: list[tuple[str, tuple[str, ...], tuple[ModelType, ...]]] = []
+
+    class FakeMigration:
+        def __init__(
+            self,
+            *,
+            tenant_id: str,
+            engine: sa.Engine,
+            apply: bool,
+            output: io.TextIOBase,
+            model_types: tuple[ModelType, ...],
+            orm_models: tuple[type[object], ...],
+        ) -> None:
+            assert engine is sqlite_engine
+            assert apply is False
+            seen_runs.append((tenant_id, tuple(model.__table__.name for model in orm_models), model_types))
+
+        def run(self) -> None:
+            return None
+
+    monkeypatch.setattr(migration_module, "Migration", FakeMigration)
+    service = migration_module.LegacyModelTypeMigrationService(
+        engine=sqlite_engine,
+        apply=False,
+        tables=("provider_models",),
+        model_types=(ModelType.LLM,),
+    )
+
+    service.migrate()
+
+    assert seen_runs == [
+        (llm_tenant_id, ("provider_models",), (ModelType.LLM,)),
+    ]
+
+
+def test_service_migrate_without_tenant_ids_discovers_all_load_balancing_tenants_for_simpler_table_scoped_query(
+    migration_module,
+    sqlite_engine: sa.Engine,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    create_minimal_legacy_model_type_schema(sqlite_engine)
+    inherit_llm_tenant_id = "00000000-0000-0000-0000-000000000511"
+    inherit_embedding_tenant_id = "00000000-0000-0000-0000-000000000522"
+    empty_tenant_id = "00000000-0000-0000-0000-000000000533"
+    for tenant_id in (inherit_llm_tenant_id, inherit_embedding_tenant_id, empty_tenant_id):
+        _insert_tenant(sqlite_engine, tenant_id=tenant_id)
+
+    created_at = datetime(2025, 1, 3, 12, 0, 0)
+    updated_at = created_at + timedelta(minutes=1)
+    _insert_load_balancing_model_config(
+        sqlite_engine,
+        row_id="40000000-0000-0000-0000-000000000511",
+        tenant_id=inherit_llm_tenant_id,
+        provider_name="openai",
+        model_name="gpt-4o-mini",
+        model_type=ModelType.LLM.value,
+        name="__inherit__",
+        encrypted_config=json.dumps({"api_key": "inherit-llm"}),
+        credential_id="50000000-0000-0000-0000-000000000511",
+        enabled=True,
+        created_at=created_at,
+        updated_at=updated_at,
+    )
+    _insert_load_balancing_model_config(
+        sqlite_engine,
+        row_id="40000000-0000-0000-0000-000000000522",
+        tenant_id=inherit_embedding_tenant_id,
+        provider_name="openai",
+        model_name="text-embedding-3-large",
+        model_type=ModelType.TEXT_EMBEDDING.value,
+        name="__inherit__",
+        encrypted_config=json.dumps({"api_key": "inherit-embedding"}),
+        credential_id="50000000-0000-0000-0000-000000000522",
+        enabled=True,
+        created_at=created_at,
+        updated_at=updated_at,
+    )
+
+    seen_runs: list[tuple[str, tuple[str, ...], tuple[ModelType, ...]]] = []
+
+    class FakeMigration:
+        def __init__(
+            self,
+            *,
+            tenant_id: str,
+            engine: sa.Engine,
+            apply: bool,
+            output: io.TextIOBase,
+            model_types: tuple[ModelType, ...],
+            orm_models: tuple[type[object], ...],
+        ) -> None:
+            assert engine is sqlite_engine
+            assert apply is False
+            seen_runs.append((tenant_id, tuple(model.__table__.name for model in orm_models), model_types))
+
+        def run(self) -> None:
+            return None
+
+    monkeypatch.setattr(migration_module, "Migration", FakeMigration)
+    # Load-balancing tenant discovery is a deliberate exception: it scans the
+    # whole table so the discovery query stays easy to understand, even when
+    # the scheduled tenant set is wider than the selected model types.
+    service = migration_module.LegacyModelTypeMigrationService(
+        engine=sqlite_engine,
+        apply=False,
+        tables=("load_balancing_model_configs",),
+        model_types=(ModelType.LLM,),
+    )
+
+    service.migrate()
+
+    assert seen_runs == [
+        (inherit_llm_tenant_id, ("load_balancing_model_configs",), (ModelType.LLM,)),
+        (inherit_embedding_tenant_id, ("load_balancing_model_configs",), (ModelType.LLM,)),
+    ]
 
 
 def test_service_migrate_with_concurrency_greater_than_one_runs_tenants_in_parallel_without_changing_migration_scope(
@@ -942,6 +1228,71 @@ def test_is_lock_timeout_error_prefers_structured_backend_codes(
     assert migration._is_lock_timeout_error(exc) is expected
 
 
+def test_process_load_balancing_model_config_row_logs_stacktrace_for_lock_timeout(
+    migration_module,
+    sqlite_engine: sa.Engine,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    output = io.StringIO()
+    migration = migration_module.Migration(
+        tenant_id="tenant-1",
+        engine=sqlite_engine,
+        apply=True,
+        output=output,
+        model_types=(ModelType.LLM,),
+        orm_models=(migration_module.LoadBalancingModelConfig,),
+    )
+    candidate = migration_module._RowWithRawModelType(
+        row=SimpleNamespace(id="lb-row-1"),
+        raw_model_type="text-generation",
+        canonical_model_type=ModelType.LLM,
+    )
+    lock_timeout_exc = OperationalError("SELECT 1", {}, SimpleNamespace(pgcode="55P03"))
+
+    class _FakeBeginContext:
+        def __enter__(self) -> None:
+            return None
+
+        def __exit__(self, exc_type, exc, tb) -> bool:
+            return False
+
+    class _FakeSession:
+        def __enter__(self):
+            return self
+
+        def __exit__(self, exc_type, exc, tb) -> bool:
+            return False
+
+        def begin(self) -> _FakeBeginContext:
+            return _FakeBeginContext()
+
+    def _fake_session_factory(engine: sa.Engine) -> _FakeSession:
+        return _FakeSession()
+
+    def _fake_reload(self, session, original_candidate, *, lock_rows: bool):
+        raise lock_timeout_exc
+
+    monkeypatch.setattr(migration_module, "_session_factory", _fake_session_factory)
+    monkeypatch.setattr(migration_module.Migration, "_configure_lock_timeout", lambda self, session: None)
+    monkeypatch.setattr(
+        migration_module.Migration,
+        "_reload_load_balancing_model_config_candidate",
+        _fake_reload,
+    )
+
+    migration._process_load_balancing_model_config_row(candidate)
+
+    lines = _parse_json_lines(output)
+    assert len(lines) == 1
+    assert lines[0]["event"] == "lock_timeout_skipped"
+    attrs = cast(dict[str, object], lines[0]["attrs"])
+    assert attrs["table_name"] == "load_balancing_model_configs"
+    assert attrs["id"] == "lb-row-1"
+    assert attrs["error"] == str(lock_timeout_exc)
+    assert isinstance(attrs["stacktrace"], str)
+    assert "OperationalError" in attrs["stacktrace"]
+
+
 def test_process_load_balancing_model_config_row_logs_update_after_sql_execution(
     migration_module,
     sqlite_engine: sa.Engine,
@@ -1024,6 +1375,41 @@ def test_process_load_balancing_model_config_row_logs_update_after_sql_execution
     ]
 
 
+def test_load_balancing_model_config_cache_delete_failure_logs_stacktrace(
+    migration_module,
+    sqlite_engine: sa.Engine,
+    dirty_fixture,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    def _raise_delete_failure(self) -> None:
+        raise RuntimeError("cache delete boom")
+
+    monkeypatch.setattr(migration_module.ProviderCredentialsCache, "delete", _raise_delete_failure)
+
+    output = io.StringIO()
+    migration_module.LegacyModelTypeMigrationService(
+        engine=sqlite_engine,
+        apply=True,
+        output=output,
+        tables=("load_balancing_model_configs",),
+        model_types=(ModelType.LLM,),
+        tenant_ids=(dirty_fixture.primary.tenant_id,),
+    ).migrate()
+
+    failed_events = [
+        cast(dict[str, object], line["attrs"])
+        for line in _parse_json_lines(output)
+        if line.get("event") == "cache_delete_failed"
+        and isinstance(line.get("attrs"), dict)
+        and cast(dict[str, object], line["attrs"]).get("table_name") == "load_balancing_model_configs"
+    ]
+
+    assert len(failed_events) == 1
+    assert failed_events[0]["error"] == "cache delete boom"
+    assert isinstance(failed_events[0]["stacktrace"], str)
+    assert "RuntimeError: cache delete boom" in cast(str, failed_events[0]["stacktrace"])
+
+
 def test_group_completed_logs_exist_for_all_grouped_tables_and_use_canonical_model_type(
     migration_module,
     sqlite_engine: sa.Engine,
@@ -1188,28 +1574,44 @@ def test_provider_model_settings_group_crossing_batches_is_completed_once_with_a
     }
 
 
-def test_load_balancing_model_configs_are_canonicalized_row_by_row_without_group_business_key_semantics(
+def test_load_balancing_inherit_rows_are_deduplicated_by_normalized_model_type_before_canonicalization(
     migration_module,
     sqlite_engine: sa.Engine,
     dirty_fixture,
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
-    inserted_row_id = "00000000-0000-0000-0000-00000000dd01"
+    older_canonical_row_id = "00000000-0000-0000-0000-00000000dd01"
+    newer_legacy_row_id = "00000000-0000-0000-0000-00000000dd02"
     created_at = datetime(2025, 1, 1, 8, 0, 0)
-    updated_at = created_at + timedelta(minutes=15)
+    older_updated_at = created_at + timedelta(minutes=15)
+    newer_updated_at = created_at + timedelta(minutes=30)
     _insert_load_balancing_model_config(
         sqlite_engine,
-        row_id=inserted_row_id,
+        row_id=older_canonical_row_id,
+        tenant_id=dirty_fixture.primary.tenant_id,
+        provider_name="openai",
+        model_name="gpt-4o-mini",
+        model_type=ModelType.LLM.value,
+        name="__inherit__",
+        encrypted_config='{"api_key":"older-inherit"}',
+        credential_id=dirty_fixture.primary.winner_credential_id,
+        enabled=True,
+        created_at=created_at,
+        updated_at=older_updated_at,
+    )
+    _insert_load_balancing_model_config(
+        sqlite_engine,
+        row_id=newer_legacy_row_id,
         tenant_id=dirty_fixture.primary.tenant_id,
         provider_name="openai",
         model_name="gpt-4o-mini",
         model_type="text-generation",
-        name=dirty_fixture.primary.loser_credential_name,
-        encrypted_config='{"api_key":"second-lb"}',
+        name="__inherit__",
+        encrypted_config='{"api_key":"newer-inherit"}',
         credential_id=dirty_fixture.primary.distinct_credential_id,
         enabled=True,
         created_at=created_at,
-        updated_at=updated_at,
+        updated_at=newer_updated_at,
     )
 
     deleted_cache_keys: list[str] = []
@@ -1221,10 +1623,7 @@ def test_load_balancing_model_configs_are_canonicalized_row_by_row_without_group
 
     tenant_id = dirty_fixture.primary.tenant_id
     table_name = "load_balancing_model_configs"
-    expected_row_ids = {
-        dirty_fixture.primary.load_balancing_config_id,
-        inserted_row_id,
-    }
+    expected_row_ids = {older_canonical_row_id, newer_legacy_row_id}
 
     dry_run_output = io.StringIO()
     migration_module.LegacyModelTypeMigrationService(
@@ -1237,44 +1636,79 @@ def test_load_balancing_model_configs_are_canonicalized_row_by_row_without_group
     ).migrate()
 
     dry_run_lines = _parse_json_lines(dry_run_output)
+    dry_run_signatures = {
+        signature
+        for signature in _collect_processing_signatures(dry_run_lines)
+        if signature[1] == table_name and signature[2] in expected_row_ids
+    }
     dry_run_row_updates = [
         cast(dict[str, object], line["attrs"])
         for line in dry_run_lines
         if line.get("event") == "row_updated"
         and isinstance(line.get("attrs"), dict)
         and cast(dict[str, object], line["attrs"]).get("table_name") == table_name
+        and str(cast(dict[str, object], line["attrs"]).get("id")) in expected_row_ids
     ]
-    assert len(dry_run_row_updates) == 2
-    assert {str(attrs["id"]) for attrs in dry_run_row_updates} == expected_row_ids
-    assert all(attrs.get("old_values") == {"model_type": "text-generation"} for attrs in dry_run_row_updates)
-    assert all(attrs.get("new_values") == {"model_type": ModelType.LLM.value} for attrs in dry_run_row_updates)
+    assert len(dry_run_row_updates) == 1
+    assert str(dry_run_row_updates[0]["id"]) == newer_legacy_row_id
+    assert dry_run_row_updates[0]["old_values"] == {"model_type": "text-generation"}
+    assert dry_run_row_updates[0]["new_values"] == {"model_type": ModelType.LLM.value}
     assert all("rewrite_source" not in attrs for attrs in dry_run_row_updates)
 
-    dry_run_group_processed = [
+    dry_run_row_deletes = [
         cast(dict[str, object], line["attrs"])
         for line in dry_run_lines
-        if line.get("event") == "group_processed"
+        if line.get("event") == "row_deleted"
         and isinstance(line.get("attrs"), dict)
         and cast(dict[str, object], line["attrs"]).get("table_name") == table_name
+        and str(cast(dict[str, object], line["attrs"]).get("id")) in expected_row_ids
     ]
-    assert dry_run_group_processed == []
+    assert len(dry_run_row_deletes) == 1
+    assert dry_run_row_deletes[0]["business_key"] == {
+        "tenant_id": tenant_id,
+        "provider_name": "openai",
+        "model_name": "gpt-4o-mini",
+        "model_type": ModelType.LLM.value,
+    }
+    assert dry_run_row_deletes[0]["merge_winner_id"] == newer_legacy_row_id
+    assert dry_run_row_deletes[0]["row"] == {
+        "id": older_canonical_row_id,
+        "tenant_id": tenant_id,
+        "provider_name": "openai",
+        "model_name": "gpt-4o-mini",
+        "model_type": ModelType.LLM.value,
+        "name": "__inherit__",
+        "encrypted_config": {"api_key": "older-inherit"},
+        "credential_id": dirty_fixture.primary.winner_credential_id,
+        "credential_source_type": CredentialSourceType.CUSTOM_MODEL.value,
+        "enabled": True,
+        "created_at": created_at.isoformat(),
+        "updated_at": older_updated_at.isoformat(),
+    }
 
-    dry_run_cache_plans = [
-        cast(dict[str, object], line["attrs"])
-        for line in dry_run_lines
-        if line.get("event") == "cache_delete_planned"
+    dry_run_deleted_index = next(
+        index
+        for index, line in enumerate(dry_run_lines)
+        if line.get("event") == "row_deleted"
         and isinstance(line.get("attrs"), dict)
-        and cast(dict[str, object], line["attrs"]).get("table_name") == table_name
-    ]
-    assert len(dry_run_cache_plans) == 2
-    assert {str(attrs["id"]) for attrs in dry_run_cache_plans} == expected_row_ids
+        and cast(dict[str, object], line["attrs"]).get("id") == older_canonical_row_id
+    )
+    dry_run_updated_index = next(
+        index
+        for index, line in enumerate(dry_run_lines)
+        if line.get("event") == "row_updated"
+        and isinstance(line.get("attrs"), dict)
+        and cast(dict[str, object], line["attrs"]).get("id") == newer_legacy_row_id
+    )
+    assert dry_run_deleted_index < dry_run_updated_index
 
-    dry_run_business_keys = [
-        _json_key(business_key)
-        for attrs in [*dry_run_row_updates, *dry_run_cache_plans]
-        if isinstance((business_key := attrs.get("business_key")), dict)
-    ]
-    assert len(set(dry_run_business_keys)) == len(dry_run_business_keys)
+    dry_run_cache_plan_ids = _cache_event_row_ids(
+        dry_run_lines,
+        table_name=table_name,
+        row_ids=expected_row_ids,
+        event_name="cache_delete_planned",
+    )
+    assert newer_legacy_row_id in dry_run_cache_plan_ids
 
     apply_output = io.StringIO()
     migration_module.LegacyModelTypeMigrationService(
@@ -1287,47 +1721,100 @@ def test_load_balancing_model_configs_are_canonicalized_row_by_row_without_group
     ).migrate()
 
     apply_lines = _parse_json_lines(apply_output)
+    apply_signatures = {
+        signature
+        for signature in _collect_processing_signatures(apply_lines)
+        if signature[1] == table_name and signature[2] in expected_row_ids
+    }
     apply_row_updates = [
         cast(dict[str, object], line["attrs"])
         for line in apply_lines
         if line.get("event") == "row_updated"
         and isinstance(line.get("attrs"), dict)
         and cast(dict[str, object], line["attrs"]).get("table_name") == table_name
+        and str(cast(dict[str, object], line["attrs"]).get("id")) in expected_row_ids
     ]
-    assert len(apply_row_updates) == 2
-    assert {str(attrs["id"]) for attrs in apply_row_updates} == expected_row_ids
+    assert len(apply_row_updates) == 1
+    assert str(apply_row_updates[0]["id"]) == newer_legacy_row_id
+    assert apply_signatures == dry_run_signatures
 
-    apply_group_processed = [
-        cast(dict[str, object], line["attrs"])
-        for line in apply_lines
-        if line.get("event") == "group_processed"
-        and isinstance(line.get("attrs"), dict)
-        and cast(dict[str, object], line["attrs"]).get("table_name") == table_name
-    ]
-    assert apply_group_processed == []
-
-    apply_cache_deletes = [
-        cast(dict[str, object], line["attrs"])
-        for line in apply_lines
-        if line.get("event") == "cache_deleted"
-        and isinstance(line.get("attrs"), dict)
-        and cast(dict[str, object], line["attrs"]).get("table_name") == table_name
-    ]
-    assert len(apply_cache_deletes) == 2
-    assert {str(attrs["id"]) for attrs in apply_cache_deletes} == expected_row_ids
-    assert len(deleted_cache_keys) == 2
-
-    apply_business_keys = [
-        _json_key(business_key)
-        for attrs in [*apply_row_updates, *apply_cache_deletes]
-        if isinstance((business_key := attrs.get("business_key")), dict)
-    ]
-    assert len(set(apply_business_keys)) == len(apply_business_keys)
+    apply_cache_delete_ids = _cache_event_row_ids(
+        apply_lines,
+        table_name=table_name,
+        row_ids=expected_row_ids,
+        event_name="cache_deleted",
+    )
+    assert apply_cache_delete_ids == dry_run_cache_plan_ids
+    assert deleted_cache_keys
 
     lb_rows = fetch_table_rows(sqlite_engine, table_name, tenant_id=tenant_id)
-    migrated_rows = [row for row in lb_rows if str(row["id"]) in expected_row_ids]
-    assert len(migrated_rows) == 2
-    assert all(row["model_type"] == ModelType.LLM.value for row in migrated_rows)
+    surviving_rows = [row for row in lb_rows if str(row["id"]) in expected_row_ids]
+    assert len(surviving_rows) == 1
+    surviving_row = surviving_rows[0]
+    assert surviving_row["id"] == newer_legacy_row_id
+    assert surviving_row["tenant_id"] == tenant_id
+    assert surviving_row["provider_name"] == "openai"
+    assert surviving_row["model_name"] == "gpt-4o-mini"
+    assert surviving_row["model_type"] == ModelType.LLM.value
+    assert surviving_row["name"] == "__inherit__"
+    assert surviving_row["encrypted_config"] == '{"api_key":"newer-inherit"}'
+    assert surviving_row["credential_id"] == dirty_fixture.primary.distinct_credential_id
+    assert surviving_row["credential_source_type"] == CredentialSourceType.CUSTOM_MODEL.value
+
+
+def test_load_balancing_non_inherit_rows_do_not_participate_in_normalized_model_type_deduplication(
+    migration_module,
+    sqlite_engine: sa.Engine,
+    dirty_fixture,
+) -> None:
+    inserted_row_id = "00000000-0000-0000-0000-00000000dd03"
+    created_at = datetime(2025, 1, 1, 8, 0, 0)
+    updated_at = created_at + timedelta(minutes=15)
+    _insert_load_balancing_model_config(
+        sqlite_engine,
+        row_id=inserted_row_id,
+        tenant_id=dirty_fixture.primary.tenant_id,
+        provider_name="openai",
+        model_name="gpt-4o-mini",
+        model_type=ModelType.LLM.value,
+        name=dirty_fixture.primary.loser_credential_name,
+        encrypted_config='{"api_key":"second-lb"}',
+        credential_id=dirty_fixture.primary.distinct_credential_id,
+        enabled=True,
+        created_at=created_at,
+        updated_at=updated_at,
+    )
+
+    output = io.StringIO()
+    migration_module.LegacyModelTypeMigrationService(
+        engine=sqlite_engine,
+        apply=True,
+        output=output,
+        tables=("load_balancing_model_configs",),
+        model_types=(ModelType.LLM,),
+        tenant_ids=(dirty_fixture.primary.tenant_id,),
+    ).migrate()
+
+    lines = _parse_json_lines(output)
+    row_deleted_events = [
+        cast(dict[str, object], line["attrs"])
+        for line in lines
+        if line.get("event") == "row_deleted"
+        and isinstance(line.get("attrs"), dict)
+        and cast(dict[str, object], line["attrs"]).get("table_name") == "load_balancing_model_configs"
+    ]
+    assert row_deleted_events == []
+
+    lb_rows = fetch_table_rows(
+        sqlite_engine,
+        "load_balancing_model_configs",
+        tenant_id=dirty_fixture.primary.tenant_id,
+    )
+    matching_rows = [
+        row for row in lb_rows if str(row["id"]) in {dirty_fixture.primary.load_balancing_config_id, inserted_row_id}
+    ]
+    assert len(matching_rows) == 2
+    assert all(row["model_type"] == ModelType.LLM.value for row in matching_rows)
 
 
 def test_migration_apply_updates_all_five_tables_and_rewrites_credential_references(