Read data-driven commissioner division ladder config

commissioners/common/ruleset_strategy/commissioner.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from collections.abc import Mapping, Sequence
 from datetime import UTC, datetime, timedelta
 from typing import Any
 
@@ -42,6 +43,9 @@
     _plural_word,
     _round_structure_description,
     _schedule_slot_description,
+    division_matches_selector,
+    select_competition_entry_division,
+    select_division_by_role,
 )
 from commissioners.common.ruleset_strategy.config import RulesetStrategyCommissionerConfig, load_image_ruleset_strategy_config
 from commissioners.common.ruleset_strategy.entrants import division_entries, select_rule
@@ -76,39 +80,36 @@ def league_migration_config(self, ctx: LeagueMigrationConfigContext) -> list[Div
 
     def migrate_league(self, ctx: LeagueMigrationContext) -> LeagueMigrationResult:
         configured_names = {division.name for division in self._config().migration_divisions}
-        competition = next((division for division in ctx.divisions if division.name == "Competition"), None)
+        desired_divisions = [division for division in ctx.divisions if division.name in configured_names]
+        fallback_competition = select_competition_entry_division(ctx.league.commissioner_config, desired_divisions)
         divisions_by_id = {division.id: division for division in ctx.divisions}
+        migration_rules = _legacy_division_migration_rules(ctx.league.commissioner_config)
         events: list[PolicyMembershipEventChange] = []
 
         for membership in ctx.memberships:
             division = divisions_by_id.get(membership.division_id)
             if division is None or division.name in configured_names:
                 continue
-            if division.name == "Dirt" and membership.status != "disqualified":
-                events.append(
-                    PolicyMembershipEventChange(
-                        league_policy_membership_id=membership.id,
-                        from_division_id=membership.division_id,
-                        to_division_id=None,
-                        status="disqualified",
-                        substatus=POLICY_MEMBERSHIP_SUBSTATUS_INACTIVE,
-                        reason="Tournament restructure Dirt->Disqualified",
-                        end_time=datetime.now(UTC),
-                        evidence=[_legacy_division_migration_evidence(division.name, "Disqualified")],
-                    )
-                )
-            elif division.name == "Wood" and competition is not None:
-                events.append(
-                    PolicyMembershipEventChange(
-                        league_policy_membership_id=membership.id,
-                        from_division_id=membership.division_id,
-                        to_division_id=competition.id,
-                        status=_membership_status(membership.status),
-                        substatus=membership.substatus,
-                        reason=f"Tournament restructure Wood->{competition.name}",
-                        evidence=[_legacy_division_migration_evidence(division.name, competition.name)],
-                    )
-                )
+            rule = next(
+                (
+                    rule
+                    for rule in migration_rules
+                    if division_matches_selector(division, rule.get("from", rule.get("match")))
+                ),
+                None,
+            )
+            if rule is None:
+                continue
+            event = _legacy_division_migration_event(
+                rule,
+                membership=membership,
+                from_division=division,
+                target_divisions=desired_divisions,
+                commissioner_config=ctx.league.commissioner_config,
+                fallback_competition=fallback_competition,
+            )
+            if event is not None:
+                events.append(event)
         return LeagueMigrationResult(policy_membership_events=events)
 
     def rank_division(self, ctx: DivisionLeaderboardContext) -> list[DivisionLeaderboardSnapshot]:
@@ -283,6 +284,138 @@ def _membership_status(status: Any) -> str:
     return status.value if hasattr(status, "value") else str(status)
 
 
+def _legacy_division_migration_rules(commissioner_config: Mapping[str, Any] | None) -> Sequence[Mapping[str, Any]]:
+    config = commissioner_config or {}
+    division_ladder = config.get("division_ladder")
+    rules = division_ladder.get("legacy_migrations") if isinstance(division_ladder, Mapping) else None
+    if isinstance(rules, Sequence) and not isinstance(rules, str):
+        return [rule for rule in rules if isinstance(rule, Mapping)]
+    return (
+        {
+            "from": {"name": "Dirt"},
+            "to": {
+                "status": "disqualified",
+                "substatus": POLICY_MEMBERSHIP_SUBSTATUS_INACTIVE,
+                "reason": "Tournament restructure Dirt->Disqualified",
+                "evidence_to": "Disqualified",
+            },
+        },
+        {
+            "from": {"name": "Wood"},
+            "to": {
+                "role": "entry",
+                "reason": "Tournament restructure Wood->{target_division_name}",
+            },
+        },
+    )
+
+
+def _legacy_division_migration_event(
+    rule: Mapping[str, Any],
+    *,
+    membership: Any,
+    from_division: Any,
+    target_divisions: Sequence[Any],
+    commissioner_config: Mapping[str, Any] | None,
+    fallback_competition: Any | None,
+) -> PolicyMembershipEventChange | None:
+    action = rule.get("to")
+    if not isinstance(action, Mapping):
+        return None
+
+    target_division = _legacy_division_migration_target(
+        action,
+        commissioner_config,
+        target_divisions,
+        fallback_competition,
+    )
+    status = action.get("status")
+    if status == "disqualified" and _membership_status(membership.status) == "disqualified":
+        return None
+    if target_division is None and status != "disqualified":
+        return None
+
+    target_name = _legacy_division_migration_target_name(action, target_division)
+    return PolicyMembershipEventChange(
+        league_policy_membership_id=membership.id,
+        from_division_id=membership.division_id,
+        to_division_id=None if target_division is None else target_division.id,
+        status=str(status) if status is not None else _membership_status(membership.status),
+        substatus=action.get("substatus", membership.substatus),
+        reason=_legacy_division_migration_reason(action, from_division.name, target_name),
+        end_time=datetime.now(UTC) if status == "disqualified" else None,
+        evidence=[_legacy_division_migration_evidence(from_division.name, target_name)],
+    )
+
+
+def _legacy_division_migration_target(
+    action: Mapping[str, Any],
+    commissioner_config: Mapping[str, Any] | None,
+    target_divisions: Sequence[Any],
+    fallback_competition: Any | None,
+) -> Any | None:
+    role = action.get("role", action.get("to_role"))
+    if isinstance(role, str):
+        target = select_division_by_role(commissioner_config, target_divisions, roles=(role,))
+        if target is not None:
+            return target
+        if role in {"entry", "default", "competition"}:
+            return fallback_competition
+
+    selector = (
+        action.get("division")
+        or action.get("to_division")
+        or action.get("division_selector")
+        or _selector_from_flat_action(action)
+    )
+    if selector is not None:
+        return next(
+            (division for division in target_divisions if division_matches_selector(division, selector)),
+            None,
+        )
+    return None
+
+
+def _selector_from_flat_action(action: Mapping[str, Any]) -> dict[str, Any] | None:
+    selector = {
+        selector_key: action[action_key]
+        for action_key, selector_key in (
+            ("division_id", "id"),
+            ("to_division_id", "id"),
+            ("division_name", "name"),
+            ("to_division_name", "name"),
+            ("division_type", "type"),
+            ("to_division_type", "type"),
+            ("division_level", "level"),
+            ("to_division_level", "level"),
+        )
+        if action_key in action
+    }
+    return selector or None
+
+
+def _legacy_division_migration_target_name(action: Mapping[str, Any], target_division: Any | None) -> str:
+    if target_division is not None:
+        return str(target_division.name)
+    evidence_to = action.get("evidence_to")
+    if evidence_to is not None:
+        return str(evidence_to)
+    status = action.get("status")
+    if status is not None:
+        return str(status).title()
+    return "Removed"
+
+
+def _legacy_division_migration_reason(action: Mapping[str, Any], from_name: str, target_name: str) -> str:
+    reason = action.get("reason")
+    if not isinstance(reason, str):
+        reason = "Tournament restructure {from_division_name}->{target_division_name}"
+    return reason.format(
+        from_division_name=from_name,
+        target_division_name=target_name,
+    )
+
+
 def _legacy_division_migration_evidence(from_division: str, to_division: str) -> PolicyMembershipEventEvidence:
     return PolicyMembershipEventEvidence(
         type="tournament_restructure",

commissioners/common/utils.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from collections import defaultdict
+from collections.abc import Mapping, Sequence
 from datetime import UTC, datetime, timedelta
 from math import ceil
 from typing import Any
@@ -36,6 +37,8 @@
 )
 AMONG_THEM_SCORE_KIND = MEAN_ROUND_SCORE_KIND
 AMONG_THEM_SCORING_MECHANICS = MEAN_SCORE_EWMA_SCORING_MECHANICS
+QUALIFIER_DIVISION_ROLES = ("qualifier", "qualifiers")
+COMPETITION_ENTRY_DIVISION_ROLES = ("entry", "default", "competition")
 
 
 def select_division(
@@ -59,8 +62,11 @@ def select_qualifier_division(
     commissioner_config: dict[str, Any] | None,
     divisions: list[DivisionSnapshot],
 ) -> DivisionSnapshot | None:
-    from commissioners.common.models import DIVISION_TYPE_STAGING
     config = commissioner_config or {}
+    division = select_division_by_role(config, divisions, roles=QUALIFIER_DIVISION_ROLES)
+    if division is not None or has_division_role_config(config, QUALIFIER_DIVISION_ROLES):
+        return division
+
     qualifiers_division_name = config.get("qualifiers_division_name")
     if not qualifiers_division_name:
         return None
@@ -78,6 +84,10 @@ def select_competition_entry_division(
 ) -> DivisionSnapshot | None:
     from commissioners.common.models import DIVISION_TYPE_COMPETITION
     config = commissioner_config or {}
+    division = select_division_by_role(config, divisions, roles=COMPETITION_ENTRY_DIVISION_ROLES)
+    if division is not None or has_division_role_config(config, COMPETITION_ENTRY_DIVISION_ROLES):
+        return division
+
     return select_division(
         divisions,
         division_name=config.get("default_division_name"),
@@ -86,6 +96,88 @@ def select_competition_entry_division(
     )
 
 
+def select_division_by_role(
+    commissioner_config: Mapping[str, Any] | None,
+    divisions: Sequence[DivisionSnapshot],
+    *,
+    roles: Sequence[str],
+) -> DivisionSnapshot | None:
+    config = commissioner_config or {}
+    return next(
+        (division for division in divisions if division_matches_any_role(division, config, roles)),
+        None,
+    )
+
+
+def has_division_role_config(commissioner_config: Mapping[str, Any] | None, roles: Sequence[str]) -> bool:
+    config = commissioner_config or {}
+    return any(_division_role_selector(config, role) is not None for role in roles)
+
+
+def division_matches_any_role(
+    division: DivisionSnapshot,
+    commissioner_config: Mapping[str, Any],
+    roles: Sequence[str],
+) -> bool:
+    return any(
+        division_matches_selector(division, _division_role_selector(commissioner_config, role)) for role in roles
+    )
+
+
+def _division_role_selector(config: Mapping[str, Any], role: str) -> Any:
+    division_roles = config.get("division_roles")
+    if isinstance(division_roles, Mapping) and role in division_roles:
+        return division_roles[role]
+
+    division_ladder = config.get("division_ladder")
+    ladder_levels: Any = None
+    if isinstance(division_ladder, Mapping):
+        ladder_levels = division_ladder.get("levels")
+    elif isinstance(division_ladder, Sequence) and not isinstance(division_ladder, str):
+        ladder_levels = division_ladder
+    if isinstance(ladder_levels, Sequence) and not isinstance(ladder_levels, str):
+        for level in ladder_levels:
+            if isinstance(level, Mapping) and level.get("role") == role:
+                return level
+    return None
+
+
+def division_matches_selector(division: DivisionSnapshot, selector: Any) -> bool:
+    if selector is None:
+        return False
+    if isinstance(selector, str):
+        return str(division.id) == selector
+    if not isinstance(selector, Mapping):
+        return False
+
+    matched = False
+    division_id = selector.get("division_id", selector.get("id"))
+    if division_id is not None:
+        matched = True
+        if str(division.id) != str(division_id):
+            return False
+
+    division_name = selector.get("division_name", selector.get("name"))
+    if division_name is not None:
+        matched = True
+        if division.name != division_name:
+            return False
+
+    division_type = selector.get("division_type", selector.get("type"))
+    if division_type is not None:
+        matched = True
+        if division.type != division_type:
+            return False
+
+    division_level = selector.get("division_level", selector.get("level"))
+    if division_level is not None:
+        matched = True
+        if division.level != division_level:
+            return False
+
+    return matched
+
+
 def division_entrants(
     memberships: list[MembershipSnapshot],
     division: DivisionSnapshot,

commissioners/ruleset_strategy_commissioner/README.md

@@ -2,10 +2,13 @@
 
 Configurable Coworld commissioner whose behavior is packaged in the container image.
 
-The runnable does not read `league.commissioner_config` for behavior. That field is a platform wire artifact and may
-contain legacy data while Coworlds roll over to container commissioners. Configs are authored in the readable
-shape below, copied into the image, and selected by the image's `RULESET_STRATEGY_CONFIG_NAME` or
-`RULESET_STRATEGY_CONFIG_PATH` environment variables.
+The runnable gets its active scheduling, seating, and transition behavior from image YAML, not from
+`league.commissioner_config`. That field is mostly a platform wire artifact and may contain legacy data while Coworlds
+roll over to container commissioners. During league migration, the runnable may read
+`division_roles`/`division_ladder.legacy_migrations` from `league.commissioner_config` to move memberships out of old
+topology tiers without baking those display names into the image. Configs are authored in the readable shape below,
+copied into the image, and selected by the image's `RULESET_STRATEGY_CONFIG_NAME` or `RULESET_STRATEGY_CONFIG_PATH`
+environment variables.
 
 The shared Dockerfile bundles configs from `configs/` and defaults to `configs/default.yaml`. Build the default baseline
 image with: