"""Durable Season & Competition Orchestration Engine.

The workflow deliberately composes existing domain services instead of
re-implementing standings, movement, qualification, draw or match rules.
Each step commits independently and is idempotent, so a failed run can resume
without duplicating business artifacts.
"""
from __future__ import annotations

from datetime import date

from django.db import transaction
from django.db.models import Q
from django.utils import timezone

from federations.constants import (
    CompetitionFormat,
    CompetitionKind,
    CompetitionScope,
    CompetitionStatus,
    FederationStatus,
    FederationType,
)
from federations.models import (
    Federation,
    FederationCompetition,
    FederationCompetitionDraw,
    FederationCompetitionGroupDraw,
    FederationLeagueAuthority,
)
from federations.qualification_service import FederationCompetitionQualificationService
from federations.services import FederationCompetitionAdministrationService
from federations.tournaments import (
    FederationTournamentDrawService,
    FederationTournamentRegistrationService,
)
from leagues.constants import (
    LEAGUE_TEAM_CAPACITY,
    LeagueCompetitionStatus,
    LeagueCompetitionType,
    LeagueMembershipStatus,
    LeagueStatus,
)
from leagues.models import League, LeagueCompetition, LeagueMembership
from leagues.season_transitions import LeagueSeasonTransitionService
from leagues.standings_services import LeagueStandingsService
from matches.services import LeagueCompetitionLifecycleService, LeagueFixtureGenerationService

from .constants import (
    SeasonCalendarPhase,
    SeasonCalendarSourceType,
    SeasonOrchestrationStatus,
    SeasonOrchestrationStepKey,
    SeasonOrchestrationStepStatus,
    SeasonStatus,
)
from .exceptions import (
    SeasonActivationGateError,
    SeasonOrchestrationConflictError,
    SeasonOrchestrationError,
    SeasonOrchestrationIntegrityError,
    SeasonOrchestrationNotReadyError,
)
from .models import (
    Season,
    SeasonCompetitionCalendarSlot,
    SeasonOrchestrationRun,
    SeasonOrchestrationStep,
)
from .orchestration_policy import (
    ORCHESTRATION_POLICY_KEY,
    ORCHESTRATION_POLICY_VERSION,
    ORCHESTRATION_SCHEMA_VERSION,
    STEP_KEYS,
    SeasonCalendarPolicyV1,
    deterministic_seed,
    payload_sha256,
)
from .services import SeasonService


class SeasonCompetitionOrchestrationService:
    """Prepare a complete successor Season from authoritative source results."""

    @classmethod
    @transaction.atomic
    def prepare(cls, *, source_season_id, target_season_id) -> SeasonOrchestrationRun:
        seasons = {
            season.pk: season
            for season in Season.objects.select_for_update()
            .filter(pk__in=[source_season_id, target_season_id])
            .order_by("pk")
        }
        try:
            source = seasons[source_season_id]
            target = seasons[target_season_id]
        except KeyError as exc:
            raise SeasonOrchestrationNotReadyError("Source or target Season does not exist.") from exc
        if source.pk == target.pk:
            raise SeasonOrchestrationNotReadyError("Source and target Seasons must be different.")

        request_payload = {
            "policy_key": ORCHESTRATION_POLICY_KEY,
            "policy_version": ORCHESTRATION_POLICY_VERSION,
            "schema_version": ORCHESTRATION_SCHEMA_VERSION,
            "source_season_id": str(source.pk),
            "target_season_id": str(target.pk),
            "steps": list(STEP_KEYS),
        }
        request_hash = payload_sha256(request_payload)
        existing = SeasonOrchestrationRun.objects.filter(
            source_season=source,
            target_season=target,
        ).first()
        if existing is not None:
            cls._verify_run_identity(existing, request_payload=request_payload, request_hash=request_hash)
            cls._ensure_step_rows(existing)
            return existing

        run = SeasonOrchestrationRun.objects.create(
            source_season=source,
            target_season=target,
            policy_key=ORCHESTRATION_POLICY_KEY,
            policy_version=ORCHESTRATION_POLICY_VERSION,
            schema_version=ORCHESTRATION_SCHEMA_VERSION,
            request_payload=request_payload,
            request_hash=request_hash,
        )
        cls._ensure_step_rows(run)
        return run

    @classmethod
    def execute(cls, *, run_id, performed_by=None) -> SeasonOrchestrationRun:
        run = cls._begin_execution(run_id=run_id)
        if run.status == SeasonOrchestrationStatus.SUCCEEDED:
            cls._verify_successful_run(run)
            return run

        handlers = {
            SeasonOrchestrationStepKey.PRECHECK: cls._step_precheck,
            SeasonOrchestrationStepKey.FINALIZE_SOURCE: cls._step_finalize_source,
            SeasonOrchestrationStepKey.COMPLETE_SOURCE: cls._step_complete_source,
            SeasonOrchestrationStepKey.MATERIALIZE_PLAYERS: cls._step_materialize_players,
            SeasonOrchestrationStepKey.APPLY_MOVEMENT: cls._step_apply_movement,
            SeasonOrchestrationStepKey.GENERATE_LEAGUES: cls._step_generate_leagues,
            SeasonOrchestrationStepKey.CREATE_COMPETITIONS: cls._step_create_competitions,
            SeasonOrchestrationStepKey.QUALIFY_COMPETITIONS: cls._step_qualify_competitions,
            SeasonOrchestrationStepKey.DRAW_COMPETITIONS: cls._step_draw_competitions,
            SeasonOrchestrationStepKey.BUILD_CALENDAR: cls._step_build_calendar,
            SeasonOrchestrationStepKey.MARK_UPCOMING: cls._step_mark_upcoming,
        }

        for step_key in STEP_KEYS:
            step = SeasonOrchestrationStep.objects.get(run_id=run.pk, step_key=step_key)
            if step.status == SeasonOrchestrationStepStatus.SUCCEEDED:
                try:
                    cls._verify_successful_step(step)
                except Exception as exc:
                    cls._mark_step_failed(step_id=step.pk, run_id=run.pk, exc=exc)
                    raise SeasonOrchestrationError(
                        f"Persisted checkpoint {step_key} failed verification: {exc}"
                    ) from exc
                continue
            try:
                cls._mark_step_running(step_id=step.pk)
                result = handlers[step_key](run_id=run.pk, performed_by=performed_by)
                cls._mark_step_succeeded(step_id=step.pk, result_payload=result)
            except Exception as exc:
                cls._mark_step_failed(step_id=step.pk, run_id=run.pk, exc=exc)
                raise SeasonOrchestrationError(
                    f"Orchestration step {step_key} failed: {exc}"
                ) from exc

        try:
            return cls._mark_run_succeeded(run_id=run.pk)
        except Exception as exc:
            cls._mark_run_failed(run_id=run.pk, exc=exc)
            raise SeasonOrchestrationError(f"Final orchestration verification failed: {exc}") from exc

    @classmethod
    @transaction.atomic
    def recover_interrupted(cls, *, run_id) -> SeasonOrchestrationRun:
        """Convert an interrupted RUNNING checkpoint into a retryable FAILED state.

        This is an explicit operator action for process crashes. Normal business
        failures are persisted automatically by execute().
        """
        try:
            run = SeasonOrchestrationRun.objects.select_for_update().get(pk=run_id)
        except SeasonOrchestrationRun.DoesNotExist as exc:
            raise SeasonOrchestrationNotReadyError("Orchestration run does not exist.") from exc
        if run.status != SeasonOrchestrationStatus.RUNNING:
            return run
        running_steps = list(
            SeasonOrchestrationStep.objects.select_for_update().filter(
                run=run, status=SeasonOrchestrationStepStatus.RUNNING
            )
        )
        now = timezone.now()
        for step in running_steps:
            step.status = SeasonOrchestrationStepStatus.FAILED
            step.failed_at = now
            step.error_code = "InterruptedExecution"
            step.error_message = "Execution was explicitly recovered after an interrupted process."
            step.save(update_fields=[
                "status", "failed_at", "error_code", "error_message", "updated_at",
            ])
        run.status = SeasonOrchestrationStatus.FAILED
        run.failed_at = now
        run.error_code = "InterruptedExecution"
        run.error_message = "Execution was explicitly recovered after an interrupted process."
        run.save(update_fields=[
            "status", "failed_at", "error_code", "error_message", "updated_at",
        ])
        return run

    @classmethod
    @transaction.atomic
    def activate_target(cls, *, run_id, today: date | None = None) -> Season:
        run = (
            SeasonOrchestrationRun.objects.select_for_update()
            .select_related("source_season", "target_season")
            .get(pk=run_id)
        )
        cls._verify_successful_run(run)
        source = Season.objects.select_for_update().get(pk=run.source_season_id)
        target = Season.objects.select_for_update().get(pk=run.target_season_id)
        today = today or timezone.localdate()

        if target.status == SeasonStatus.ACTIVE:
            return target
        if target.status != SeasonStatus.UPCOMING:
            raise SeasonActivationGateError("Only the prepared UPCOMING target Season may be activated.")
        if today < target.start_date:
            raise SeasonActivationGateError(
                f"Target Season cannot activate before {target.start_date.isoformat()}."
            )
        if today > target.end_date:
            raise SeasonActivationGateError("Target Season end date has already passed.")

        if source.status == SeasonStatus.COMPLETED:
            SeasonService.archive(season_id=source.pk)
        elif source.status != SeasonStatus.ARCHIVED:
            raise SeasonActivationGateError("Source Season must be COMPLETED or ARCHIVED before activation.")
        return SeasonService.activate(season_id=target.pk)

    # ------------------------------------------------------------------
    # execution/checkpoint state
    # ------------------------------------------------------------------
    @classmethod
    @transaction.atomic
    def _begin_execution(cls, *, run_id):
        try:
            run = SeasonOrchestrationRun.objects.select_for_update().get(pk=run_id)
        except SeasonOrchestrationRun.DoesNotExist as exc:
            raise SeasonOrchestrationNotReadyError("Orchestration run does not exist.") from exc
        if run.status == SeasonOrchestrationStatus.SUCCEEDED:
            return run
        if run.status == SeasonOrchestrationStatus.RUNNING:
            raise SeasonOrchestrationConflictError("This orchestration run is already executing.")
        run.status = SeasonOrchestrationStatus.RUNNING
        run.current_step = ""
        run.attempt_count += 1
        run.started_at = run.started_at or timezone.now()
        run.failed_at = None
        run.error_code = ""
        run.error_message = ""
        run.save(
            update_fields=[
                "status", "current_step", "attempt_count", "started_at",
                "failed_at", "error_code", "error_message", "updated_at",
            ]
        )
        return run

    @classmethod
    @transaction.atomic
    def _mark_step_running(cls, *, step_id):
        step = SeasonOrchestrationStep.objects.select_for_update().select_related("run").get(pk=step_id)
        if step.status == SeasonOrchestrationStepStatus.SUCCEEDED:
            return step
        step.status = SeasonOrchestrationStepStatus.RUNNING
        step.attempt_count += 1
        step.started_at = timezone.now()
        step.completed_at = None
        step.failed_at = None
        step.error_code = ""
        step.error_message = ""
        step.save(update_fields=[
            "status", "attempt_count", "started_at", "completed_at", "failed_at",
            "error_code", "error_message", "updated_at",
        ])
        run = SeasonOrchestrationRun.objects.select_for_update().get(pk=step.run_id)
        run.current_step = step.step_key
        run.save(update_fields=["current_step", "updated_at"])
        return step

    @classmethod
    @transaction.atomic
    def _mark_step_succeeded(cls, *, step_id, result_payload):
        step = SeasonOrchestrationStep.objects.select_for_update().get(pk=step_id)
        result_hash = payload_sha256(result_payload)
        step.status = SeasonOrchestrationStepStatus.SUCCEEDED
        step.result_payload = result_payload
        step.result_hash = result_hash
        step.completed_at = timezone.now()
        step.failed_at = None
        step.error_code = ""
        step.error_message = ""
        step.save(update_fields=[
            "status", "result_payload", "result_hash", "completed_at", "failed_at",
            "error_code", "error_message", "updated_at",
        ])

    @classmethod
    @transaction.atomic
    def _mark_step_failed(cls, *, step_id, run_id, exc):
        now = timezone.now()
        step = SeasonOrchestrationStep.objects.select_for_update().get(pk=step_id)
        step.status = SeasonOrchestrationStepStatus.FAILED
        step.failed_at = now
        step.error_code = exc.__class__.__name__[:120]
        step.error_message = str(exc)[:1000]
        step.save(update_fields=[
            "status", "failed_at", "error_code", "error_message", "updated_at",
        ])
        run = SeasonOrchestrationRun.objects.select_for_update().get(pk=run_id)
        run.status = SeasonOrchestrationStatus.FAILED
        run.failed_at = now
        run.error_code = step.error_code
        run.error_message = step.error_message
        run.save(update_fields=["status", "failed_at", "error_code", "error_message", "updated_at"])

    @classmethod
    @transaction.atomic
    def _mark_run_failed(cls, *, run_id, exc):
        run = SeasonOrchestrationRun.objects.select_for_update().get(pk=run_id)
        run.status = SeasonOrchestrationStatus.FAILED
        run.failed_at = timezone.now()
        run.error_code = exc.__class__.__name__[:120]
        run.error_message = str(exc)[:1000]
        run.save(update_fields=[
            "status", "failed_at", "error_code", "error_message", "updated_at",
        ])
        return run

    @classmethod
    @transaction.atomic
    def _mark_run_succeeded(cls, *, run_id):
        run = SeasonOrchestrationRun.objects.select_for_update().get(pk=run_id)
        steps = list(run.steps.order_by("ordinal"))
        if len(steps) != len(STEP_KEYS) or any(
            step.status != SeasonOrchestrationStepStatus.SUCCEEDED for step in steps
        ):
            raise SeasonOrchestrationIntegrityError("A run cannot succeed before every step succeeds.")
        result_payload = {
            "policy_key": run.policy_key,
            "policy_version": run.policy_version,
            "source_season_id": str(run.source_season_id),
            "target_season_id": str(run.target_season_id),
            "step_result_hashes": [step.result_hash for step in steps],
            "calendar_slot_count": run.calendar_slots.count(),
        }
        run.status = SeasonOrchestrationStatus.SUCCEEDED
        run.current_step = ""
        run.result_payload = result_payload
        run.result_hash = payload_sha256(result_payload)
        run.completed_at = timezone.now()
        run.failed_at = None
        run.error_code = ""
        run.error_message = ""
        run.save(update_fields=[
            "status", "current_step", "result_payload", "result_hash", "completed_at",
            "failed_at", "error_code", "error_message", "updated_at",
        ])
        return run

    @classmethod
    def _ensure_step_rows(cls, run):
        for ordinal, step_key in enumerate(STEP_KEYS, start=1):
            input_payload = {
                "run_id": str(run.pk),
                "step_key": step_key,
                "ordinal": ordinal,
                "source_season_id": str(run.source_season_id),
                "target_season_id": str(run.target_season_id),
                "policy_version": run.policy_version,
            }
            input_hash = payload_sha256(input_payload)
            step, created = SeasonOrchestrationStep.objects.get_or_create(
                run=run,
                step_key=step_key,
                defaults={
                    "ordinal": ordinal,
                    "input_payload": input_payload,
                    "input_hash": input_hash,
                },
            )
            if not created and (
                step.ordinal != ordinal
                or step.input_payload != input_payload
                or step.input_hash != input_hash
            ):
                raise SeasonOrchestrationIntegrityError("Persisted orchestration step identity changed.")

    @staticmethod
    def _verify_run_identity(run, *, request_payload, request_hash):
        if (
            run.policy_key != ORCHESTRATION_POLICY_KEY
            or run.policy_version != ORCHESTRATION_POLICY_VERSION
            or run.schema_version != ORCHESTRATION_SCHEMA_VERSION
            or run.request_payload != request_payload
            or run.request_hash != request_hash
            or payload_sha256(run.request_payload) != run.request_hash
        ):
            raise SeasonOrchestrationIntegrityError("Persisted orchestration request evidence is inconsistent.")

    @staticmethod
    def _verify_successful_step(step):
        if not step.result_hash or payload_sha256(step.result_payload) != step.result_hash:
            raise SeasonOrchestrationIntegrityError(
                f"Persisted result evidence for step {step.step_key} is corrupted."
            )

    @classmethod
    def _verify_successful_run(cls, run):
        if run.status != SeasonOrchestrationStatus.SUCCEEDED:
            raise SeasonActivationGateError("Target Season has not completed orchestration.")
        if not run.result_hash or payload_sha256(run.result_payload) != run.result_hash:
            raise SeasonOrchestrationIntegrityError("Persisted orchestration result evidence is corrupted.")
        steps = list(run.steps.order_by("ordinal"))
        if len(steps) != len(STEP_KEYS):
            raise SeasonOrchestrationIntegrityError("Persisted orchestration step set is incomplete.")
        for step in steps:
            if step.status != SeasonOrchestrationStepStatus.SUCCEEDED:
                raise SeasonOrchestrationIntegrityError("Successful run contains a non-successful step.")
            cls._verify_successful_step(step)

    # ------------------------------------------------------------------
    # business steps
    # ------------------------------------------------------------------
    @classmethod
    def _step_precheck(cls, *, run_id, performed_by=None):
        run = cls._run_with_seasons(run_id)
        source, target = run.source_season, run.target_season
        if source.status != SeasonStatus.ACTIVE:
            raise SeasonOrchestrationNotReadyError("Initial orchestration requires an ACTIVE source Season.")
        if target.status != SeasonStatus.DRAFT:
            raise SeasonOrchestrationNotReadyError("Initial orchestration requires a DRAFT target Season.")
        if target.start_date <= source.end_date:
            raise SeasonOrchestrationNotReadyError("Target Season must start after the source Season ends.")
        if target.year <= source.year:
            raise SeasonOrchestrationNotReadyError("Target Season year must be later than source Season year.")

        mothers = list(Federation.objects.filter(federation_type=FederationType.MOTHER, status=FederationStatus.ACTIVE))
        if len(mothers) != 1:
            raise SeasonOrchestrationNotReadyError("Exactly one ACTIVE mother federation is required.")
        mother = mothers[0]
        regions = list(
            Federation.objects.filter(
                federation_type=FederationType.REGIONAL,
                status=FederationStatus.ACTIVE,
                parent=mother,
            ).order_by("jurisdiction", "pk")
        )
        if not regions:
            raise SeasonOrchestrationNotReadyError("At least one ACTIVE regional federation is required.")
        from federations.models import ContinentalAllocation
        from federations.allocations import validate_capacity
        allocations = list(ContinentalAllocation.objects.filter(
            mother_federation=mother, season=target, published_at__isnull=False,
        ))
        policy = getattr(target, "automation", None)
        if policy and policy.continental_enabled and not any(a.kind == CompetitionKind.CHAMPIONS_LEAGUE for a in allocations):
            raise SeasonOrchestrationNotReadyError("Mother federation must publish continental seats for the next season, or explicitly disable continental play in its calendar.")
        for allocation in allocations:
            validate_capacity(sum(row["seats"] for row in allocation.seats))
        kinds = {allocation.kind for allocation in allocations}
        if CompetitionKind.CHAMPIONS_LEAGUE_2 in kinds and CompetitionKind.CHAMPIONS_LEAGUE not in kinds:
            raise SeasonOrchestrationNotReadyError("Publish Champions League allocation before Champions League 2.")

        active_leagues = list(League.objects.filter(status=LeagueStatus.ACTIVE, competitions__season=source).distinct().order_by("country", "competition_type", "level"))
        if not active_leagues:
            raise SeasonOrchestrationNotReadyError("No ACTIVE leagues exist for orchestration.")
        authority_by_league = {
            row.league_id: row
            for row in FederationLeagueAuthority.objects.select_related("federation").filter(
                league_id__in=[league.pk for league in active_leagues]
            )
        }
        league_summary = []
        for league in active_leagues:
            authority = authority_by_league.get(league.pk)
            if (
                authority is None
                or authority.federation.parent_id != mother.pk
                or authority.federation.federation_type != FederationType.REGIONAL
                or authority.federation.status != FederationStatus.ACTIVE
            ):
                raise SeasonOrchestrationNotReadyError(
                    f"League {league.pk} is not bound to an ACTIVE regional federation tree."
                )
            if authority.federation.jurisdiction and (
                authority.federation.jurisdiction.strip().casefold() != league.country.strip().casefold()
            ):
                raise SeasonOrchestrationNotReadyError(
                    f"League {league.name} country does not match its federation jurisdiction."
                )
            try:
                competition = LeagueCompetition.objects.get(league=league, season=source)
            except LeagueCompetition.DoesNotExist as exc:
                raise SeasonOrchestrationNotReadyError(
                    f"Source Season is missing competition execution for league {league.name}."
                ) from exc
            if competition.status not in {LeagueCompetitionStatus.ACTIVE, LeagueCompetitionStatus.COMPLETED}:
                raise SeasonOrchestrationNotReadyError(
                    f"League {league.name} must be ACTIVE or COMPLETED before season closure."
                )
            member_count = LeagueMembership.objects.filter(
                league=league,
                season=source,
                status=LeagueMembershipStatus.REGISTERED,
            ).count()
            if member_count != LEAGUE_TEAM_CAPACITY:
                raise SeasonOrchestrationNotReadyError(
                    f"League {league.name} must contain exactly {LEAGUE_TEAM_CAPACITY} registered participants."
                )
            league_summary.append({
                "league_id": str(league.pk),
                "competition_id": str(competition.pk),
                "country": league.country,
                "competition_type": league.competition_type,
                "level": league.level,
            })

        for allocation in allocations:
            for row in allocation.seats:
                if not any(str(authority_by_league[league.pk].federation_id) == row["regional_federation_id"]
                           and league.level == 1 and league.competition_type == LeagueCompetitionType.ADULT
                           for league in active_leagues):
                    raise SeasonOrchestrationNotReadyError("Allocated federation has no participating level-1 adult league.")

        nonterminal_federation = FederationCompetition.objects.filter(season=source).exclude(
            status__in=[CompetitionStatus.COMPLETED, CompetitionStatus.CANCELLED]
        )
        if nonterminal_federation.exists():
            raise SeasonOrchestrationNotReadyError(
                "Every source federation competition must be COMPLETED or CANCELLED before season closure."
            )

        dirty_target = (
            LeagueMembership.objects.filter(season=target).exists()
            or LeagueCompetition.objects.filter(season=target).exists()
            or FederationCompetition.objects.filter(season=target).exists()
        )
        if dirty_target:
            raise SeasonOrchestrationNotReadyError(
                "Target Season must be clean before orchestration creates memberships and competitions."
            )

        return {
            "mother_federation_id": str(mother.pk),
            "regional_federation_ids": [str(region.pk) for region in regions],
            "active_league_count": len(active_leagues),
            "leagues": league_summary,
        }

    @classmethod
    def _step_finalize_source(cls, *, run_id, performed_by=None):
        run = cls._run_with_seasons(run_id)
        source = run.source_season
        competitions = list(
            LeagueCompetition.objects.select_related("league")
            .filter(season=source, league__status=LeagueStatus.ACTIVE)
            .order_by("league__country", "league__competition_type", "league__level")
        )
        if not competitions:
            raise SeasonOrchestrationNotReadyError("No source league competitions are available to finalize.")
        finalized = []
        for competition in competitions:
            if competition.status == LeagueCompetitionStatus.ACTIVE:
                competition = LeagueCompetitionLifecycleService.complete(competition_id=competition.pk)
            elif competition.status == LeagueCompetitionStatus.COMPLETED:
                LeagueStandingsService.finalize(competition_id=competition.pk)
            else:
                raise SeasonOrchestrationNotReadyError(
                    f"League competition {competition.pk} is not ready for finalization."
                )
            snapshot = competition.standings_snapshot
            finalized.append({
                "competition_id": str(competition.pk),
                "league_id": str(competition.league_id),
                "standings_hash": snapshot.table_hash,
            })
        return {"finalized_count": len(finalized), "competitions": finalized}

    @classmethod
    def _step_complete_source(cls, *, run_id, performed_by=None):
        run = cls._run_with_seasons(run_id)
        source = Season.objects.get(pk=run.source_season_id)
        if source.status == SeasonStatus.ACTIVE:
            source = SeasonService.complete(season_id=source.pk)
        elif source.status not in {SeasonStatus.COMPLETED, SeasonStatus.ARCHIVED}:
            raise SeasonOrchestrationNotReadyError("Source Season cannot be completed from its current state.")
        return {
            "source_season_id": str(source.pk),
            "status": source.status,
            "completed_at": source.completed_at.isoformat() if source.completed_at else None,
        }

    @classmethod
    def _step_materialize_players(cls, *, run_id, performed_by=None):
        from training.progression_services import PlayerSeasonAgingService

        run = cls._run_with_seasons(run_id)
        results = PlayerSeasonAgingService.process_season(season_id=run.source_season_id)
        return {
            "source_season_id": str(run.source_season_id),
            "processed_players": len(results),
            "declined_players": sum(1 for row in results if row.applied_decline_points > 0),
            "retirement_eligible_players": sum(1 for row in results if row.retirement_eligible),
            "aging_result_hashes": [row.evidence_hash for row in results],
        }

    @classmethod
    def _step_apply_movement(cls, *, run_id, performed_by=None):
        run = cls._run_with_seasons(run_id)
        groups = list(
            League.objects.filter(status=LeagueStatus.ACTIVE, competitions__season=run.source_season)
            .order_by("country", "competition_type")
            .values_list("country", "competition_type")
            .distinct()
        )
        transitions = []
        for country, competition_type in groups:
            transition = LeagueSeasonTransitionService.apply(
                source_season_id=run.source_season_id,
                target_season_id=run.target_season_id,
                country=country,
                competition_type=competition_type,
                performed_by=performed_by,
            )
            transitions.append({
                "transition_id": str(transition.pk),
                "country": transition.country,
                "competition_type": transition.competition_type,
                "transition_hash": transition.transition_hash,
            })
        return {"transition_count": len(transitions), "transitions": transitions}

    @classmethod
    def _step_generate_leagues(cls, *, run_id, performed_by=None):
        run = cls._run_with_seasons(run_id)
        generated = []
        for league in League.objects.filter(status=LeagueStatus.ACTIVE, memberships__season=run.target_season, memberships__status="REGISTERED").distinct().order_by("country", "competition_type", "level"):
            competition = LeagueFixtureGenerationService.generate(
                league_id=league.pk,
                season_id=run.target_season_id,
            )
            generated.append({
                "league_id": str(league.pk),
                "competition_id": str(competition.pk),
                "status": competition.status,
                "fixture_count": competition.fixtures.count(),
            })
        return {"league_competition_count": len(generated), "competitions": generated}

    @classmethod
    def _step_create_competitions(cls, *, run_id, performed_by=None):
        run = cls._run_with_seasons(run_id)
        target = run.target_season
        mother = cls._mother()
        created = []
        for region, competition_type in cls._cup_specs(target):
            suffix = "National Cup" if competition_type == LeagueCompetitionType.ADULT else "Youth Cup"
            competition = cls._get_or_create_federation_competition(
                federation=region, target=target, kind=CompetitionKind.REGIONAL_CUP,
                competition_type=competition_type, name=f"{region.jurisdiction or region.name} {suffix} {target.year}",
                format=CompetitionFormat.SINGLE_ELIMINATION, scope=CompetitionScope.REGIONAL,
                performed_by=performed_by,
            )
            created.append(str(competition.pk))
        from federations.models import ContinentalAllocation
        kinds = set(ContinentalAllocation.objects.filter(season=target, mother_federation=mother,
                    published_at__isnull=False).values_list("kind", flat=True))
        for kind, label in (
            (CompetitionKind.CHAMPIONS_LEAGUE, "Champions League"),
            (CompetitionKind.CHAMPIONS_LEAGUE_2, "Champions League 2"),
        ):
            if kind not in kinds:
                continue
            competition = cls._get_or_create_federation_competition(
                federation=mother,
                target=target,
                kind=kind,
                competition_type=LeagueCompetitionType.ADULT,
                name=f"FTG {label} {target.year}",
                format=CompetitionFormat.GROUP_STAGE_KNOCKOUT,
                scope=CompetitionScope.GLOBAL,
                performed_by=performed_by,
            )
            created.append(str(competition.pk))
        return {"competition_count": len(created), "competition_ids": created}

    @classmethod
    def _step_qualify_competitions(cls, *, run_id, performed_by=None):
        run = cls._run_with_seasons(run_id)
        mother = cls._mother()
        results = {}
        higher = None
        for kind in (CompetitionKind.CHAMPIONS_LEAGUE, CompetitionKind.CHAMPIONS_LEAGUE_2):
            competition = FederationCompetition.objects.filter(
                federation=mother, season=run.target_season, kind=kind,
            ).first()
            if competition is None:
                continue
            qualified = FederationCompetitionQualificationService.apply(
                competition_id=competition.pk, source_season_id=run.source_season_id,
                higher_competition_id=higher if kind == CompetitionKind.CHAMPIONS_LEAGUE_2 else None,
                performed_by=performed_by,
            )
            higher = competition.pk
            results[kind.lower()] = {"competition_id": str(competition.pk),
                "qualification_run_id": str(qualified.pk), "qualified_count": qualified.qualified_count,
                "qualification_hash": qualified.qualification_hash}
        return results

    @classmethod
    def _step_draw_competitions(cls, *, run_id, performed_by=None):
        run = cls._run_with_seasons(run_id)
        target = run.target_season
        competitions = cls._managed_target_competitions(target=target)
        results = []
        for competition in competitions:
            if competition.kind == CompetitionKind.REGIONAL_CUP:
                participants = FederationTournamentRegistrationService.register_all_regional_eligible(
                    competition_id=competition.pk,
                    performed_by=performed_by,
                )
                if len(participants) < 2:
                    raise SeasonOrchestrationNotReadyError(
                        f"Competition {competition.name} has fewer than two eligible participants."
                    )
            seed = deterministic_seed(run.pk, competition.pk, "draw")
            draw = FederationTournamentDrawService.draw(
                competition_id=competition.pk,
                seed=seed,
                performed_by=performed_by,
            )
            results.append({
                "competition_id": str(competition.pk),
                "kind": competition.kind,
                "competition_type": competition.competition_type,
                "draw_id": str(draw.pk),
                "seed": seed,
                "participant_count": draw.participant_count,
            })
        return {"draw_count": len(results), "draws": results}

    @classmethod
    def _step_build_calendar(cls, *, run_id, performed_by=None):
        run = cls._run_with_seasons(run_id)
        target = run.target_season
        created_or_verified = []

        league_competitions = list(
            LeagueCompetition.objects.select_related("league")
            .filter(season=target)
            .order_by("league__country", "league__competition_type", "league__level")
        )
        for competition in league_competitions:
            policy = getattr(target, "automation", None)
            windows = SeasonCalendarPolicyV1.league_windows(
                start_date=target.start_date,
                end_date=target.end_date,
                round_count=38,
                round_gap_days=policy.round_gap_days if policy else None,
                break_start=timezone.localtime(policy.midseason_opens_at).date() if policy else None,
                break_end=timezone.localtime(policy.midseason_closes_at).date() if policy else None,
            )
            for window in windows:
                slot = cls._persist_calendar_slot(
                    run=run,
                    season=target,
                    source_type=SeasonCalendarSourceType.LEAGUE,
                    league_competition=competition,
                    federation_competition=None,
                    window=window,
                )
                created_or_verified.append(str(slot.pk))

        federation_competitions = cls._managed_target_competitions(target=target)
        continental_index = 0
        for competition in federation_competitions:
            if competition.format == CompetitionFormat.SINGLE_ELIMINATION:
                try:
                    draw = FederationCompetitionDraw.objects.get(competition=competition)
                except FederationCompetitionDraw.DoesNotExist as exc:
                    raise SeasonOrchestrationIntegrityError(
                        f"Competition {competition.pk} is missing knockout draw evidence."
                    ) from exc
                windows = SeasonCalendarPolicyV1.regional_cup_windows(
                    start_date=target.start_date,
                    end_date=target.end_date,
                    bracket_size=draw.bracket_size,
                )
            elif competition.format == CompetitionFormat.GROUP_STAGE_KNOCKOUT:
                try:
                    draw = FederationCompetitionGroupDraw.objects.get(competition=competition)
                except FederationCompetitionGroupDraw.DoesNotExist as exc:
                    raise SeasonOrchestrationIntegrityError(
                        f"Competition {competition.pk} is missing group draw evidence."
                    ) from exc
                windows = SeasonCalendarPolicyV1.group_stage_knockout_windows(
                    start_date=target.start_date,
                    end_date=target.end_date,
                    group_count=draw.group_count,
                    offset_days=continental_index * 7,
                )
                continental_index += 1
            else:
                raise SeasonOrchestrationIntegrityError("Unsupported competition format in calendar builder.")
            for window in windows:
                slot = cls._persist_calendar_slot(
                    run=run,
                    season=target,
                    source_type=SeasonCalendarSourceType.FEDERATION,
                    league_competition=None,
                    federation_competition=competition,
                    window=window,
                )
                created_or_verified.append(str(slot.pk))

        return {
            "calendar_policy_key": SeasonCalendarPolicyV1.policy_key,
            "calendar_policy_version": SeasonCalendarPolicyV1.policy_version,
            "slot_count": len(created_or_verified),
            "slot_ids": created_or_verified,
            "kickoff_semantics": "date_window_only",
        }

    @classmethod
    def _step_mark_upcoming(cls, *, run_id, performed_by=None):
        run = cls._run_with_seasons(run_id)
        target = Season.objects.get(pk=run.target_season_id)
        if target.status == SeasonStatus.DRAFT:
            target = SeasonService.mark_upcoming(season_id=target.pk)
        elif target.status != SeasonStatus.UPCOMING:
            raise SeasonOrchestrationNotReadyError("Target Season cannot become UPCOMING from its current state.")
        return {"target_season_id": str(target.pk), "status": target.status}

    # ------------------------------------------------------------------
    # helpers
    # ------------------------------------------------------------------
    @staticmethod
    def _run_with_seasons(run_id):
        return SeasonOrchestrationRun.objects.select_related("source_season", "target_season").get(pk=run_id)

    @staticmethod
    def _mother():
        mothers = list(Federation.objects.filter(federation_type=FederationType.MOTHER, status=FederationStatus.ACTIVE))
        if len(mothers) != 1:
            raise SeasonOrchestrationNotReadyError("Exactly one ACTIVE mother federation is required.")
        return mothers[0]

    @staticmethod
    def _regions(*, mother):
        regions = list(
            Federation.objects.filter(
                parent=mother,
                federation_type=FederationType.REGIONAL,
                status=FederationStatus.ACTIVE,
            ).order_by("jurisdiction", "pk")
        )
        if not regions:
            raise SeasonOrchestrationNotReadyError("No ACTIVE regional federations exist.")
        return regions

    @classmethod
    def _get_or_create_federation_competition(
        cls,
        *,
        federation,
        target,
        kind,
        competition_type,
        name,
        format,
        scope,
        performed_by,
    ):
        existing = list(
            FederationCompetition.objects.filter(
                federation=federation,
                season=target,
                kind=kind,
                competition_type=competition_type,
            ).order_by("pk")
        )
        if len(existing) > 1:
            raise SeasonOrchestrationIntegrityError("Target Season contains duplicate managed competitions.")
        if existing:
            competition = existing[0]
            if competition.format != format or competition.scope != scope:
                raise SeasonOrchestrationIntegrityError("Existing target competition shape differs from orchestration policy.")
            return competition
        return FederationCompetitionAdministrationService.create_competition(
            federation_id=federation.pk,
            name=name,
            kind=kind,
            scope=scope,
            season_id=target.pk,
            competition_type=competition_type,
            format=format,
            performed_by=performed_by,
        )

    @classmethod
    def _managed_target_competitions(cls, *, target):
        mother = cls._mother()
        from federations.models import ContinentalAllocation
        specs = cls._cup_specs(target)
        kinds = list(ContinentalAllocation.objects.filter(mother_federation=mother, season=target,
                     published_at__isnull=False).values_list("kind", flat=True))
        rows = list(FederationCompetition.objects.select_related("federation").filter(season=target)
            .filter(Q(kind=CompetitionKind.REGIONAL_CUP, federation_id__in=[region.pk for region, _ in specs])
                    | Q(federation=mother, kind__in=kinds))
            .order_by("scope", "federation_id", "competition_type", "kind", "pk"))
        expected = len(specs) + len(kinds)
        if len(rows) != expected:
            raise SeasonOrchestrationIntegrityError(f"Expected {expected} managed competitions, found {len(rows)}.")
        return rows

    @classmethod
    def _cup_specs(cls, target):
        specs = []
        mother = cls._mother()
        for region in cls._regions(mother=mother):
            for competition_type in (LeagueCompetitionType.ADULT, LeagueCompetitionType.YOUTH):
                if LeagueCompetition.objects.filter(season=target, league__federation_authority__federation=region,
                                                     league__competition_type=competition_type).exists():
                    specs.append((region, competition_type))
        return specs

    @staticmethod
    def _single_target_competition(*, federation, season, kind, competition_type):
        rows = list(
            FederationCompetition.objects.filter(
                federation=federation,
                season=season,
                kind=kind,
                competition_type=competition_type,
            ).order_by("pk")
        )
        if len(rows) != 1:
            raise SeasonOrchestrationIntegrityError("Expected exactly one managed target competition.")
        return rows[0]

    @staticmethod
    def _persist_calendar_slot(
        *,
        run,
        season,
        source_type,
        league_competition,
        federation_competition,
        window,
    ):
        evidence = {
            "calendar_policy_key": SeasonCalendarPolicyV1.policy_key,
            "calendar_policy_version": SeasonCalendarPolicyV1.policy_version,
            "run_id": str(run.pk) if run else None,
            "season_id": str(season.pk),
            "source_type": source_type,
            "league_competition_id": str(league_competition.pk) if league_competition else None,
            "federation_competition_id": str(federation_competition.pk) if federation_competition else None,
            **window.to_payload(),
        }
        evidence_hash = payload_sha256(evidence)
        lookup = {
            "run": run,
            "phase": window.phase,
            "sequence": window.sequence,
        }
        if source_type == SeasonCalendarSourceType.LEAGUE:
            lookup["league_competition"] = league_competition
        else:
            lookup["federation_competition"] = federation_competition
        existing = SeasonCompetitionCalendarSlot.objects.filter(**lookup).first()
        if existing is not None:
            if (
                existing.season_id != season.pk
                or existing.source_type != source_type
                or existing.scheduled_date != window.scheduled_date
                or existing.metadata != window.metadata
                or existing.evidence_hash != evidence_hash
            ):
                raise SeasonOrchestrationIntegrityError("Persisted calendar slot differs from deterministic policy.")
            return existing
        return SeasonCompetitionCalendarSlot.objects.create(
            run=run,
            season=season,
            source_type=source_type,
            league_competition=league_competition,
            federation_competition=federation_competition,
            phase=window.phase,
            sequence=window.sequence,
            scheduled_date=window.scheduled_date,
            metadata=window.metadata,
            evidence_hash=evidence_hash,
        )