# #region DatasetReviewOrchestrator [C:5] [TYPE Module] [SEMANTICS pydantic, dataset, review, orchestration, session]
# @BRIEF Coordinate dataset review session startup and lifecycle-safe intake recovery for one authenticated user.
# @LAYER Domain
# @RELATION DEPENDS_ON -> [DatasetReviewSessionRepository]
# @RELATION DEPENDS_ON -> [SemanticSourceResolver]
# @RELATION DEPENDS_ON -> [SupersetContextExtractor]
# @RELATION DEPENDS_ON -> [SupersetCompilationAdapter]
# @RELATION DEPENDS_ON -> [TaskManager]
# @RELATION DISPATCHES -> [OrchestratorHelpers]
# @RELATION DISPATCHES -> [OrchestratorCommands]
# @PRE  session mutations must execute inside a persisted session boundary scoped to one authenticated user.
# @POST  state transitions are persisted atomically and emit observable progress for long-running steps.
# @SIDE_EFFECT  creates task records, updates session aggregates, triggers upstream Superset calls, persists audit artifacts.
# @DATA_CONTRACT  Input[SessionCommand] -> Output[DatasetReviewSession | CompiledPreview | DatasetRunContext]
# @INVARIANT  Launch is blocked unless a current session has no open blocking findings, all launch-sensitive mappings are approved, and a non-stale Superset-generated compiled preview matches the current input fingerprint.
# @RATIONALE  Original 1198-line monolith violated INV_7 (400-line module limit). Decomposed into commands and helpers sub-modules while preserving the orchestrator class as the single entry point.
# @REJECTED  Keeping all orchestration logic in one file because it exceeded the fractal limit by 3x.

from __future__ import annotations

from datetime import UTC, datetime
from typing import Any, cast

from src.core.config_manager import ConfigManager
from src.core.logger import belief_scope, logger
from src.core.task_manager import TaskManager
from src.core.utils.superset_compilation_adapter import (
    PreviewCompilationPayload,
    SqlLabLaunchPayload,
    SupersetCompilationAdapter,
)
from src.core.utils.superset_context_extractor import (
    SupersetContextExtractor,
    SupersetParsedContext,
)
from src.models.dataset_review import (
    ApprovalState,
    DatasetReviewSession,
    DatasetRunContext,
    ExecutionMapping,
    FilterConfidenceState,
    FilterRecoveryStatus,
    FilterSource,
    FindingArea,
    FindingSeverity,
    ImportedFilter,
    LaunchStatus,
    MappingMethod,
    MappingStatus,
    PreviewStatus,
    ReadinessState,
    RecommendedAction,
    ResolutionState,
    SessionPhase,
    SessionStatus,
    TemplateVariable,
    ValidationFinding,
    VariableKind,
)
from src.services.dataset_review.event_logger import SessionEventPayload
from src.services.dataset_review.orchestrator_pkg._commands import (
    LaunchDatasetCommand,
    LaunchDatasetResult,
    PreparePreviewCommand,
    PreparePreviewResult,
    StartSessionCommand,
    StartSessionResult,
)
from src.services.dataset_review.orchestrator_pkg._helpers import (
    build_execution_snapshot,
    build_initial_profile,
    build_launch_blockers,
    build_partial_recovery_findings,
    get_latest_preview,
    parse_dataset_selection,
)
from src.services.dataset_review.repositories.session_repository import (
    DatasetReviewSessionRepository,
)
from src.services.dataset_review.semantic_resolver import SemanticSourceResolver

logger = cast(Any, logger)


# #region DatasetReviewOrchestrator [C:5] [TYPE Class]
# @BRIEF Coordinate safe session startup while preserving cross-user isolation and explicit partial recovery.
# @RELATION DEPENDS_ON -> [DatasetReviewSessionRepository]
# @RELATION DEPENDS_ON -> [SupersetContextExtractor]
# @RELATION DEPENDS_ON -> [TaskManager]
# @RELATION DEPENDS_ON -> [ConfigManager]
# @RELATION DEPENDS_ON -> [SemanticSourceResolver]
# @RELATION CALLS -> [OrchestratorHelpers]
# @PRE  constructor dependencies are valid and tied to the current request/task scope.
# @POST  orchestrator instance can execute session-scoped mutations for one authenticated user.
# @SIDE_EFFECT  downstream operations may persist session/profile/finding state and enqueue background tasks.
# @DATA_CONTRACT  Input[StartSessionCommand] -> Output[StartSessionResult]
# @INVARIANT  session ownership is preserved on every mutation and recovery remains explicit when partial.
class DatasetReviewOrchestrator:
    # region DatasetReviewOrchestrator_init [TYPE Function]
    # @PURPOSE: Bind repository, config, and task dependencies required by the orchestration boundary.
    # @PRE  repository/config_manager are valid collaborators for the current request scope.
    # @POST  Instance holds collaborator references used by start/preview/launch orchestration methods.
    def __init__(
        self,
        repository: DatasetReviewSessionRepository,
        config_manager: ConfigManager,
        task_manager: TaskManager | None = None,
        semantic_resolver: SemanticSourceResolver | None = None,
    ) -> None:
        self.repository = repository
        self.config_manager = config_manager
        self.task_manager = task_manager
        self.semantic_resolver = semantic_resolver or SemanticSourceResolver()

    # endregion DatasetReviewOrchestrator_init

    # region start_session [TYPE Function]
    # @PURPOSE: Initialize a new session from a Superset link or dataset selection and trigger context recovery.
    # @RELATION CALLS -> [EXT:method:SupersetContextExtractor.parse_superset_link]
    # @RELATION CALLS -> [EXT:method:TaskManager.create_task]
    # @PRE  source input is non-empty and environment is accessible.
    # @POST  session exists in persisted storage with intake/recovery state and task linkage when async work is required.
    # @SIDE_EFFECT  persists session and may enqueue recovery task.
    # @DATA_CONTRACT  Input[StartSessionCommand] -> Output[StartSessionResult]
    # @INVARIANT  no cross-user session leakage occurs; session and follow-up task remain owned by the authenticated user.
    def start_session(self, command: StartSessionCommand) -> StartSessionResult:
        with belief_scope("DatasetReviewOrchestrator.start_session"):
            normalized_source_kind = str(command.source_kind or "").strip()
            normalized_source_input = str(command.source_input or "").strip()
            normalized_environment_id = str(command.environment_id or "").strip()

            if not normalized_source_input:
                logger.explore("Blocked dataset review session start due to empty source input")
                raise ValueError("source_input must be non-empty")

            if normalized_source_kind not in {"superset_link", "dataset_selection"}:
                logger.explore("Blocked dataset review session start due to unsupported source kind", extra={"source_kind": normalized_source_kind})
                raise ValueError("source_kind must be 'superset_link' or 'dataset_selection'")

            environment = self.config_manager.get_environment(normalized_environment_id)
            if environment is None:
                logger.explore("Blocked dataset review session start because environment was not found", extra={"environment_id": normalized_environment_id})
                raise ValueError("Environment not found")

            logger.reason("Starting dataset review session", extra={"user_id": command.user.id, "environment_id": normalized_environment_id, "source_kind": normalized_source_kind})

            parsed_context: SupersetParsedContext | None = None
            findings: list[ValidationFinding] = []
            dataset_ref = normalized_source_input
            dataset_id: int | None = None
            dashboard_id: int | None = None
            readiness_state = ReadinessState.IMPORTING
            recommended_action = RecommendedAction.REVIEW_DOCUMENTATION
            current_phase = SessionPhase.RECOVERY

            if normalized_source_kind == "superset_link":
                extractor = SupersetContextExtractor(environment)
                parsed_context = extractor.parse_superset_link(normalized_source_input)
                dataset_ref = parsed_context.dataset_ref
                dataset_id = parsed_context.dataset_id
                dashboard_id = parsed_context.dashboard_id

                if parsed_context.partial_recovery:
                    readiness_state = ReadinessState.RECOVERY_REQUIRED
                    recommended_action = RecommendedAction.REVIEW_DOCUMENTATION
                    findings.extend(build_partial_recovery_findings(parsed_context))
                else:
                    readiness_state = ReadinessState.REVIEW_READY
            else:
                dataset_ref, dataset_id = parse_dataset_selection(normalized_source_input)
                readiness_state = ReadinessState.REVIEW_READY
                current_phase = SessionPhase.REVIEW

            session = DatasetReviewSession(
                user_id=command.user.id,
                environment_id=normalized_environment_id,
                source_kind=normalized_source_kind,
                source_input=normalized_source_input,
                dataset_ref=dataset_ref,
                dataset_id=dataset_id,
                dashboard_id=dashboard_id,
                readiness_state=readiness_state,
                recommended_action=recommended_action,
                status=SessionStatus.ACTIVE,
                current_phase=current_phase,
            )
            persisted_session = cast(Any, self.repository.create_session(session))

            recovered_filters: list[ImportedFilter] = []
            template_variables: list[TemplateVariable] = []
            execution_mappings: list[ExecutionMapping] = []
            if normalized_source_kind == "superset_link" and parsed_context is not None:
                recovered_filters, template_variables, execution_mappings, findings = (
                    self._build_recovery_bootstrap(
                        environment=environment,
                        session=persisted_session,
                        parsed_context=parsed_context,
                        findings=findings,
                    )
                )

            profile = build_initial_profile(
                session_id=persisted_session.session_id,
                parsed_context=parsed_context,
                dataset_ref=dataset_ref,
            )
            self.repository.event_logger.log_event(
                SessionEventPayload(
                    session_id=persisted_session.session_id,
                    actor_user_id=command.user.id,
                    event_type="session_started",
                    event_summary="Dataset review session shell created",
                    current_phase=persisted_session.current_phase.value,
                    readiness_state=persisted_session.readiness_state.value,
                    event_details={
                        "source_kind": persisted_session.source_kind,
                        "dataset_ref": persisted_session.dataset_ref,
                        "dataset_id": persisted_session.dataset_id,
                        "dashboard_id": persisted_session.dashboard_id,
                        "partial_recovery": bool(parsed_context and parsed_context.partial_recovery),
                    },
                )
            )
            persisted_session = self.repository.save_profile_and_findings(
                persisted_session.session_id,
                command.user.id,
                profile,
                findings,
            )
            if recovered_filters or template_variables or execution_mappings:
                persisted_session = self.repository.save_recovery_state(
                    persisted_session.session_id,
                    command.user.id,
                    recovered_filters,
                    template_variables,
                    execution_mappings,
                )

            active_task_id = self._enqueue_recovery_task(
                command=command,
                session=persisted_session,
                parsed_context=parsed_context,
            )
            if active_task_id:
                persisted_session.active_task_id = active_task_id
                self.repository.bump_session_version(persisted_session)
                self.repository.db.commit()
                self.repository.db.refresh(persisted_session)
                self.repository.event_logger.log_event(
                    SessionEventPayload(
                        session_id=persisted_session.session_id,
                        actor_user_id=command.user.id,
                        event_type="recovery_task_linked",
                        event_summary="Recovery task linked to dataset review session",
                        current_phase=persisted_session.current_phase.value,
                        readiness_state=persisted_session.readiness_state.value,
                        event_details={"task_id": active_task_id},
                    )
                )
                logger.reason("Linked recovery task to started dataset review session", extra={"session_id": persisted_session.session_id, "task_id": active_task_id})

            logger.reflect("Dataset review session start completed", extra={"session_id": persisted_session.session_id, "dataset_ref": persisted_session.dataset_ref, "readiness_state": persisted_session.readiness_state.value, "active_task_id": persisted_session.active_task_id, "finding_count": len(findings)})
            return StartSessionResult(
                session=persisted_session,
                parsed_context=parsed_context,
                findings=findings,
            )

    # endregion start_session

    # region prepare_launch_preview [TYPE Function]
    # @PURPOSE: Assemble effective execution inputs and trigger Superset-side preview compilation.
    # @RELATION CALLS -> [SupersetCompilationAdapter.compile_preview]
    # @PRE  all required variables have candidate values or explicitly accepted defaults.
    # @POST  returns preview artifact in pending, ready, failed, or stale state.
    # @SIDE_EFFECT  persists preview attempt and upstream compilation diagnostics.
    # @DATA_CONTRACT  Input[PreparePreviewCommand] -> Output[PreparePreviewResult]
    def prepare_launch_preview(self, command: PreparePreviewCommand) -> PreparePreviewResult:
        with belief_scope("DatasetReviewOrchestrator.prepare_launch_preview"):
            session = self.repository.load_session_detail(command.session_id, command.user.id)
            if session is None or session.user_id != command.user.id:
                logger.explore("Preview preparation rejected because owned session was not found", extra={"session_id": command.session_id, "user_id": command.user.id})
                raise ValueError("Session not found")

            if command.expected_version is not None:
                self.repository.require_session_version(session, command.expected_version)

            if session.dataset_id is None:
                raise ValueError("Preview requires a resolved dataset_id")

            environment = self.config_manager.get_environment(session.environment_id)
            if environment is None:
                raise ValueError("Environment not found")

            execution_snapshot = build_execution_snapshot(session)
            preview_blockers = execution_snapshot["preview_blockers"]
            if preview_blockers:
                logger.explore("Preview preparation blocked by incomplete execution context", extra={"session_id": session.session_id, "blocked_reasons": preview_blockers})
                raise ValueError("Preview blocked: " + "; ".join(preview_blockers))

            adapter = SupersetCompilationAdapter(environment)
            preview = adapter.compile_preview(
                PreviewCompilationPayload(
                    session_id=session.session_id,
                    dataset_id=session.dataset_id,
                    preview_fingerprint=execution_snapshot["preview_fingerprint"],
                    template_params=execution_snapshot["template_params"],
                    effective_filters=execution_snapshot["effective_filters"],
                )
            )
            persisted_preview = self.repository.save_preview(
                session.session_id,
                command.user.id,
                preview,
                expected_version=command.expected_version,
            )

            session.current_phase = SessionPhase.PREVIEW
            session.last_activity_at = datetime.now(UTC)
            if persisted_preview.preview_status == PreviewStatus.READY:
                launch_blockers = build_launch_blockers(session=session, execution_snapshot=execution_snapshot, preview=persisted_preview)
                if launch_blockers:
                    session.readiness_state = ReadinessState.COMPILED_PREVIEW_READY
                    session.recommended_action = RecommendedAction.APPROVE_MAPPING
                else:
                    session.readiness_state = ReadinessState.RUN_READY
                    session.recommended_action = RecommendedAction.LAUNCH_DATASET
            else:
                session.readiness_state = ReadinessState.PARTIALLY_READY
                session.recommended_action = RecommendedAction.GENERATE_SQL_PREVIEW
            self.repository.db.commit()
            self.repository.db.refresh(session)
            self.repository.event_logger.log_event(
                SessionEventPayload(
                    session_id=session.session_id,
                    actor_user_id=command.user.id,
                    event_type="preview_generated",
                    event_summary="Superset preview generation persisted",
                    current_phase=session.current_phase.value,
                    readiness_state=session.readiness_state.value,
                    event_details={"preview_id": persisted_preview.preview_id, "preview_status": persisted_preview.preview_status.value, "preview_fingerprint": persisted_preview.preview_fingerprint},
                )
            )

            logger.reflect("Superset preview preparation completed", extra={"session_id": session.session_id, "preview_id": persisted_preview.preview_id, "preview_status": persisted_preview.preview_status.value})
            return PreparePreviewResult(session=session, preview=persisted_preview, blocked_reasons=[])

    # endregion prepare_launch_preview

    # region launch_dataset [TYPE Function]
    # @PURPOSE: Start the approved dataset execution through SQL Lab and persist run context for audit/replay.
    # @RELATION CALLS -> [SupersetCompilationAdapter.create_sql_lab_session]
    # @PRE  session is run-ready and compiled preview is current.
    # @POST  returns persisted run context with SQL Lab session reference and launch outcome.
    # @SIDE_EFFECT  creates SQL Lab execution session and audit snapshot.
    # @DATA_CONTRACT  Input[LaunchDatasetCommand] -> Output[LaunchDatasetResult]
    # @INVARIANT  launch remains blocked unless blocking findings are closed, approvals are satisfied, and the latest preview fingerprint matches current execution inputs.
    def launch_dataset(self, command: LaunchDatasetCommand) -> LaunchDatasetResult:
        with belief_scope("DatasetReviewOrchestrator.launch_dataset"):
            session = self.repository.load_session_detail(command.session_id, command.user.id)
            if session is None or session.user_id != command.user.id:
                logger.explore("Launch rejected because owned session was not found", extra={"session_id": command.session_id, "user_id": command.user.id})
                raise ValueError("Session not found")

            if command.expected_version is not None:
                self.repository.require_session_version(session, command.expected_version)

            if session.dataset_id is None:
                raise ValueError("Launch requires a resolved dataset_id")

            environment = self.config_manager.get_environment(session.environment_id)
            if environment is None:
                raise ValueError("Environment not found")

            execution_snapshot = build_execution_snapshot(session)
            current_preview = get_latest_preview(session)
            launch_blockers_list = build_launch_blockers(session=session, execution_snapshot=execution_snapshot, preview=current_preview)
            if launch_blockers_list:
                logger.explore("Launch gate blocked dataset execution", extra={"session_id": session.session_id, "blocked_reasons": launch_blockers_list})
                raise ValueError("Launch blocked: " + "; ".join(launch_blockers_list))

            adapter = SupersetCompilationAdapter(environment)
            try:
                sql_lab_session_ref = adapter.create_sql_lab_session(
                    SqlLabLaunchPayload(
                        session_id=session.session_id,
                        dataset_id=session.dataset_id,
                        preview_id=current_preview.preview_id,
                        compiled_sql=str(current_preview.compiled_sql or ""),
                        template_params=execution_snapshot["template_params"],
                    )
                )
                launch_status = LaunchStatus.STARTED
                launch_error = None
            except Exception as exc:
                logger.explore("SQL Lab launch failed after passing gates", extra={"session_id": session.session_id, "error": str(exc)})
                sql_lab_session_ref = "unavailable"
                launch_status = LaunchStatus.FAILED
                launch_error = str(exc)

            run_context = DatasetRunContext(
                session_id=session.session_id,
                dataset_ref=session.dataset_ref,
                environment_id=session.environment_id,
                preview_id=current_preview.preview_id,
                sql_lab_session_ref=sql_lab_session_ref,
                effective_filters=execution_snapshot["effective_filters"],
                template_params=execution_snapshot["template_params"],
                approved_mapping_ids=execution_snapshot["approved_mapping_ids"],
                semantic_decision_refs=execution_snapshot["semantic_decision_refs"],
                open_warning_refs=execution_snapshot["open_warning_refs"],
                launch_status=launch_status,
                launch_error=launch_error,
            )
            persisted_run_context = self.repository.save_run_context(
                session.session_id,
                command.user.id,
                run_context,
                expected_version=command.expected_version,
            )

            session.current_phase = SessionPhase.LAUNCH
            session.last_activity_at = datetime.now(UTC)
            if launch_status == LaunchStatus.FAILED:
                session.readiness_state = ReadinessState.COMPILED_PREVIEW_READY
                session.recommended_action = RecommendedAction.LAUNCH_DATASET
            else:
                session.readiness_state = ReadinessState.RUN_IN_PROGRESS
                session.recommended_action = RecommendedAction.EXPORT_OUTPUTS
            self.repository.db.commit()
            self.repository.db.refresh(session)
            self.repository.event_logger.log_event(
                SessionEventPayload(
                    session_id=session.session_id,
                    actor_user_id=command.user.id,
                    event_type="dataset_launch_requested",
                    event_summary="Dataset launch handoff persisted",
                    current_phase=session.current_phase.value,
                    readiness_state=session.readiness_state.value,
                    event_details={"run_context_id": persisted_run_context.run_context_id, "launch_status": persisted_run_context.launch_status.value, "preview_id": persisted_run_context.preview_id, "sql_lab_session_ref": persisted_run_context.sql_lab_session_ref},
                )
            )

            logger.reflect("Dataset launch orchestration completed with audited run context", extra={"session_id": session.session_id, "run_context_id": persisted_run_context.run_context_id, "launch_status": persisted_run_context.launch_status.value})
            return LaunchDatasetResult(session=session, run_context=persisted_run_context, blocked_reasons=[])

    # endregion launch_dataset

    # region _build_recovery_bootstrap [TYPE Function]
    # @PURPOSE: Recover and materialize initial imported filters, template variables, and draft execution mappings after session creation.
    # @PRE  session belongs to the just-created review aggregate and parsed_context was produced for the same environment scope.
    # @POST  Returns bootstrap imported filters, template variables, execution mappings, and updated findings without persisting them directly.
    # @SIDE_EFFECT  Performs Superset reads through the extractor and may append warning findings for incomplete recovery.
    def _build_recovery_bootstrap(
        self,
        environment,
        session: DatasetReviewSession,
        parsed_context: SupersetParsedContext,
        findings: list[ValidationFinding],
    ) -> tuple[list[ImportedFilter], list[TemplateVariable], list[ExecutionMapping], list[ValidationFinding]]:
        session_record = cast(Any, session)
        extractor = SupersetContextExtractor(environment)
        imported_filters_payload = extractor.recover_imported_filters(parsed_context)
        if imported_filters_payload is None:
            imported_filters_payload = []
        imported_filters = [
            ImportedFilter(
                session_id=session_record.session_id,
                filter_name=str(item.get("filter_name") or f"imported_filter_{index}"),
                display_name=item.get("display_name"),
                raw_value=item.get("raw_value"),
                raw_value_masked=bool(item.get("raw_value_masked", False)),
                normalized_value=item.get("normalized_value"),
                source=FilterSource(str(item.get("source") or FilterSource.SUPERSET_URL.value)),
                confidence_state=FilterConfidenceState(str(item.get("confidence_state") or FilterConfidenceState.UNRESOLVED.value)),
                requires_confirmation=bool(item.get("requires_confirmation", False)),
                recovery_status=FilterRecoveryStatus(str(item.get("recovery_status") or FilterRecoveryStatus.PARTIAL.value)),
                notes=item.get("notes"),
            )
            for index, item in enumerate(imported_filters_payload)
        ]

        template_variables: list[TemplateVariable] = []
        execution_mappings: list[ExecutionMapping] = []

        if session.dataset_id is not None:
            try:
                dataset_payload = parsed_context.dataset_payload
                if not isinstance(dataset_payload, dict):
                    dataset_payload = extractor.client.get_dataset_detail(session_record.dataset_id)
                discovered_variables = extractor.discover_template_variables(dataset_payload)
                template_variables = [
                    TemplateVariable(
                        session_id=session_record.session_id,
                        variable_name=str(item.get("variable_name") or f"variable_{index}"),
                        expression_source=str(item.get("expression_source") or ""),
                        variable_kind=VariableKind(str(item.get("variable_kind") or VariableKind.UNKNOWN.value)),
                        is_required=bool(item.get("is_required", True)),
                        default_value=item.get("default_value"),
                        mapping_status=MappingStatus(str(item.get("mapping_status") or MappingStatus.UNMAPPED.value)),
                    )
                    for index, item in enumerate(discovered_variables)
                ]
            except Exception as exc:
                if "dataset_template_variable_discovery_failed" not in parsed_context.unresolved_references:
                    parsed_context.unresolved_references.append("dataset_template_variable_discovery_failed")
                if not any(f.caused_by_ref == "dataset_template_variable_discovery_failed" for f in findings):
                    findings.append(
                        ValidationFinding(
                            area=FindingArea.TEMPLATE_MAPPING,
                            severity=FindingSeverity.WARNING,
                            code="TEMPLATE_VARIABLE_DISCOVERY_FAILED",
                            title="Template variables could not be discovered",
                            message="Session remains usable, but dataset template variables still need review.",
                            resolution_state=ResolutionState.OPEN,
                            caused_by_ref="dataset_template_variable_discovery_failed",
                        )
                    )
                logger.explore("Template variable discovery failed during session bootstrap", extra={"session_id": session_record.session_id, "dataset_id": session_record.dataset_id, "error": str(exc)})

        filter_lookup = {str(f.filter_name or "").strip().lower(): f for f in imported_filters if str(f.filter_name or "").strip()}
        for tv in template_variables:
            matched_filter = filter_lookup.get(str(tv.variable_name or "").strip().lower())
            if matched_filter is None:
                continue
            requires_explicit_approval = bool(matched_filter.requires_confirmation or matched_filter.recovery_status != FilterRecoveryStatus.RECOVERED)
            execution_mappings.append(
                ExecutionMapping(
                    session_id=session_record.session_id,
                    filter_id=matched_filter.filter_id,
                    variable_id=tv.variable_id,
                    mapping_method=MappingMethod.DIRECT_MATCH,
                    raw_input_value=matched_filter.raw_value,
                    effective_value=matched_filter.normalized_value if matched_filter.normalized_value is not None else matched_filter.raw_value,
                    transformation_note="Bootstrapped from Superset recovery context",
                    warning_level=None,
                    requires_explicit_approval=requires_explicit_approval,
                    approval_state=ApprovalState.PENDING if requires_explicit_approval else ApprovalState.NOT_REQUIRED,
                    approved_by_user_id=None,
                    approved_at=None,
                )
            )

        return imported_filters, template_variables, execution_mappings, findings

    # endregion _build_recovery_bootstrap

    # region _enqueue_recovery_task [TYPE Function]
    # @PURPOSE: Link session start to observable async recovery when task infrastructure is available.
    # @PRE  session is already persisted.
    # @POST  returns task identifier when a task could be enqueued, otherwise None.
    # @SIDE_EFFECT  may create one background task for progressive recovery.
    def _enqueue_recovery_task(
        self,
        command: StartSessionCommand,
        session: DatasetReviewSession,
        parsed_context: SupersetParsedContext | None,
    ) -> str | None:
        session_record = cast(Any, session)
        if self.task_manager is None:
            logger.reason("Dataset review session started without task manager; continuing synchronously", extra={"session_id": session_record.session_id})
            return None

        task_params: dict[str, Any] = {
            "session_id": session_record.session_id,
            "user_id": command.user.id,
            "environment_id": session_record.environment_id,
            "source_kind": session_record.source_kind,
            "source_input": session_record.source_input,
            "dataset_ref": session_record.dataset_ref,
            "dataset_id": session_record.dataset_id,
            "dashboard_id": session_record.dashboard_id,
            "partial_recovery": bool(parsed_context and parsed_context.partial_recovery),
        }

        create_task = getattr(self.task_manager, "create_task", None)
        if create_task is None:
            logger.explore("Task manager has no create_task method; skipping recovery enqueue")
            return None

        try:
            task_object = create_task(plugin_id="dataset-review-recovery", params=task_params)
        except TypeError:
            logger.explore("Recovery task enqueue skipped because task manager create_task contract is incompatible", extra={"session_id": session_record.session_id})
            return None

        task_id = getattr(task_object, "id", None)
        return str(task_id) if task_id else None

    # endregion _enqueue_recovery_task


# #endregion DatasetReviewOrchestrator

# #endregion DatasetReviewOrchestrator
