# #region DatasetReviewRoutes [C:3] [TYPE Module] [SEMANTICS fastapi, dataset, review, feedback, route]
# @BRIEF Persist thumbs up/down feedback for clarification question/answer content.


from __future__ import annotations

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

from fastapi import APIRouter, Depends, HTTPException, Query, Response, status

from src.api.routes.dataset_review_pkg._dependencies import (
    BatchApproveMappingRequest,
    BatchApproveSemanticRequest,
    ClarificationAnswerRequest,
    ClarificationAnswerResultResponse,
    ClarificationStateResponse,
    ExportArtifactResponse,
    FeedbackRequest,
    FeedbackResponse,
    FieldSemanticUpdateRequest,
    LaunchDatasetResponse,
    MappingCollectionResponse,
    PreviewEnqueueResultResponse,
    SessionCollectionResponse,
    StartSessionRequest,
    UpdateExecutionMappingRequest,
    UpdateSessionRequest,
    _build_documentation_export,
    _build_session_version_conflict_http_exception,
    _build_sql_lab_redirect_url,
    _build_validation_export,
    _commit_owned_session_mutation,
    _get_clarification_engine,
    _get_latest_clarification_session_or_404,
    _get_orchestrator,
    _get_owned_field_or_404,
    _get_owned_mapping_or_404,
    _get_owned_session_or_404,
    _get_repository,
    _prepare_owned_session_mutation,
    _record_session_event,
    _require_auto_review_flag,
    _require_clarification_flag,
    _require_execution_flag,
    _require_session_version_header,
    _serialize_clarification_state,
    _serialize_empty_clarification_state,
    _serialize_execution_mapping,
    _serialize_preview,
    _serialize_run_context,
    _serialize_semantic_field,
    _serialize_session_detail,
    _serialize_session_summary,
    _update_semantic_field_state,
)
from src.core.logger import belief_scope, logger
from src.dependencies import get_config_manager, get_current_user, has_permission
from src.models.auth import User
from src.models.dataset_review import (
    ApprovalState,
    ArtifactFormat,
    FieldProvenance,
    MappingMethod,
    PreviewStatus,
    ReadinessState,
    RecommendedAction,
    SessionStatus,
)
from src.schemas.dataset_review import (
    CompiledPreviewDto,
    ExecutionMappingDto,
    SemanticFieldEntryDto,
    SessionDetail,
    SessionSummary,
    ValidationFindingDto,
)
from src.services.dataset_review.clarification_engine import (
    ClarificationAnswerCommand,
    ClarificationStateResult,
)
from src.services.dataset_review.orchestrator import (
    LaunchDatasetCommand,
    PreparePreviewCommand,
    StartSessionCommand,
)
from src.services.dataset_review.repositories.session_repository import (
    DatasetReviewSessionVersionConflictError,
)

router = APIRouter(prefix="/api/dataset-orchestration", tags=["Dataset Orchestration"])


# #region list_sessions [C:3] [TYPE Function]
# @BRIEF List resumable dataset review sessions for the current user.
@router.get(
    "/sessions",
    response_model=SessionCollectionResponse,
    dependencies=[
        Depends(_require_auto_review_flag),
        Depends(has_permission("dataset:session", "READ")),
    ],
)
async def list_sessions(
    page: int = Query(1, ge=1),
    page_size: int = Query(20, ge=1, le=100),
    repository=Depends(_get_repository),
    current_user: User = Depends(get_current_user),
):
    with belief_scope("dataset_review.list_sessions"):
        logger.reason(
            "Listing dataset review sessions",
            extra={"user_id": current_user.id, "page": page, "page_size": page_size},
        )
        sessions = repository.list_sessions_for_user(current_user.id)
        start = (page - 1) * page_size
        end = start + page_size
        items = [_serialize_session_summary(s) for s in sessions[start:end]]
        logger.reflect(
            "Session page assembled",
            extra={
                "user_id": current_user.id,
                "returned": len(items),
                "total": len(sessions),
            },
        )
        return SessionCollectionResponse(
            items=items,
            total=len(sessions),
            page=page,
            page_size=page_size,
            has_next=end < len(sessions),
        )


# #endregion list_sessions


# #region start_session [C:4] [TYPE Function]
# @BRIEF Start a new dataset review session from a Superset link or dataset selection.
@router.post(
    "/sessions",
    response_model=SessionSummary,
    status_code=status.HTTP_201_CREATED,
    dependencies=[
        Depends(_require_auto_review_flag),
        Depends(has_permission("dataset:session", "MANAGE")),
    ],
)
async def start_session(
    request: StartSessionRequest,
    orchestrator=Depends(_get_orchestrator),
    current_user: User = Depends(get_current_user),
):
    with belief_scope("start_session"):
        logger.reason(
            "Starting dataset review session",
            extra={
                "user_id": current_user.id,
                "environment_id": request.environment_id,
            },
        )
        try:
            result = orchestrator.start_session(
                StartSessionCommand(
                    user=current_user,
                    environment_id=request.environment_id,
                    source_kind=request.source_kind,
                    source_input=request.source_input,
                )
            )
        except ValueError as exc:
            logger.explore(
                "Session start rejected",
                extra={"user_id": current_user.id, "error": str(exc)},
            )
            detail = str(exc)
            sc = (
                status.HTTP_404_NOT_FOUND
                if detail == "Environment not found"
                else status.HTTP_400_BAD_REQUEST
            )
            raise HTTPException(status_code=sc, detail=detail) from exc
        logger.reflect(
            "Session started", extra={"session_id": result.session.session_id}
        )
        return _serialize_session_summary(result.session)


# #endregion start_session


# #region get_session_detail [C:3] [TYPE Function]
# @BRIEF Return the full accessible dataset review session aggregate.
@router.get(
    "/sessions/{session_id}",
    response_model=SessionDetail,
    dependencies=[
        Depends(_require_auto_review_flag),
        Depends(has_permission("dataset:session", "READ")),
    ],
)
async def get_session_detail(
    session_id: str,
    repository=Depends(_get_repository),
    current_user: User = Depends(get_current_user),
):
    with belief_scope("dataset_review.get_session_detail"):
        session = _get_owned_session_or_404(repository, session_id, current_user)
        return _serialize_session_detail(session)


# #endregion get_session_detail


# #region update_session [C:3] [TYPE Function]
# @BRIEF Update resumable lifecycle status for an owned session.
@router.patch(
    "/sessions/{session_id}",
    response_model=SessionSummary,
    dependencies=[
        Depends(_require_auto_review_flag),
        Depends(has_permission("dataset:session", "MANAGE")),
    ],
)
async def update_session(
    session_id: str,
    request: UpdateSessionRequest,
    session_version: int = Depends(_require_session_version_header),
    repository=Depends(_get_repository),
    current_user: User = Depends(get_current_user),
):
    with belief_scope("update_session"):
        session = _prepare_owned_session_mutation(
            repository, session_id, current_user, session_version
        )
        session_record = cast(Any, session)
        session_record.status = request.status
        if request.status == SessionStatus.PAUSED:
            session_record.recommended_action = RecommendedAction.RESUME_SESSION
        elif request.status in {
            SessionStatus.ARCHIVED,
            SessionStatus.CANCELLED,
            SessionStatus.COMPLETED,
        }:
            session_record.active_task_id = None
        _commit_owned_session_mutation(repository, session)
        _record_session_event(
            repository,
            session,
            current_user,
            event_type="session_status_updated",
            event_summary="Dataset review session lifecycle updated",
            event_details={
                "status": session_record.status.value,
                "version": session_record.version,
            },
        )
        return _serialize_session_summary(session)


# #endregion update_session


# #region delete_session [C:3] [TYPE Function]
# @BRIEF Archive or hard-delete a session owned by the current user.
@router.delete(
    "/sessions/{session_id}",
    status_code=status.HTTP_204_NO_CONTENT,
    dependencies=[
        Depends(_require_auto_review_flag),
        Depends(has_permission("dataset:session", "MANAGE")),
    ],
)
async def delete_session(
    session_id: str,
    hard_delete: bool = Query(False),
    session_version: int = Depends(_require_session_version_header),
    repository=Depends(_get_repository),
    current_user: User = Depends(get_current_user),
):
    with belief_scope("delete_session"):
        session = _prepare_owned_session_mutation(
            repository, session_id, current_user, session_version
        )
        if hard_delete:
            _record_session_event(
                repository,
                session,
                current_user,
                event_type="session_deleted",
                event_summary="Session hard-deleted",
                event_details={"hard_delete": True},
            )
            repository.db.delete(session)
            repository.db.commit()
            return Response(status_code=status.HTTP_204_NO_CONTENT)
        session_record = cast(Any, session)
        session_record.status = SessionStatus.ARCHIVED
        session_record.active_task_id = None
        _commit_owned_session_mutation(repository, session)
        _record_session_event(
            repository,
            session,
            current_user,
            event_type="session_archived",
            event_summary="Session archived",
            event_details={"hard_delete": False, "version": session_record.version},
        )
        return Response(status_code=status.HTTP_204_NO_CONTENT)


# #endregion delete_session


# #region export_documentation [C:3] [TYPE Function]
# @BRIEF Export documentation output for the current session.
@router.get(
    "/sessions/{session_id}/exports/documentation",
    response_model=ExportArtifactResponse,
    dependencies=[
        Depends(_require_auto_review_flag),
        Depends(has_permission("dataset:session", "READ")),
    ],
)
async def export_documentation(
    session_id: str,
    format: ArtifactFormat = Query(ArtifactFormat.JSON),
    repository=Depends(_get_repository),
    current_user: User = Depends(get_current_user),
):
    with belief_scope("export_documentation"):
        if format not in {ArtifactFormat.JSON, ArtifactFormat.MARKDOWN}:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="Only json and markdown exports are supported",
            )
        session = _get_owned_session_or_404(repository, session_id, current_user)
        payload = _build_documentation_export(session, format)
        return ExportArtifactResponse(
            artifact_id=f"documentation-{session.session_id}-{format.value}",
            session_id=session.session_id,
            artifact_type="documentation",
            format=format.value,
            storage_ref=payload["storage_ref"],
            created_by_user_id=current_user.id,
            content=payload["content"],
        )


# #endregion export_documentation


# #region export_validation [C:3] [TYPE Function]
# @BRIEF Export validation findings for the current session.
@router.get(
    "/sessions/{session_id}/exports/validation",
    response_model=ExportArtifactResponse,
    dependencies=[
        Depends(_require_auto_review_flag),
        Depends(has_permission("dataset:session", "READ")),
    ],
)
async def export_validation(
    session_id: str,
    format: ArtifactFormat = Query(ArtifactFormat.JSON),
    repository=Depends(_get_repository),
    current_user: User = Depends(get_current_user),
):
    with belief_scope("export_validation"):
        if format not in {ArtifactFormat.JSON, ArtifactFormat.MARKDOWN}:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="Only json and markdown exports are supported",
            )
        session = _get_owned_session_or_404(repository, session_id, current_user)
        payload = _build_validation_export(session, format)
        return ExportArtifactResponse(
            artifact_id=f"validation-{session.session_id}-{format.value}",
            session_id=session.session_id,
            artifact_type="validation_report",
            format=format.value,
            storage_ref=payload["storage_ref"],
            created_by_user_id=current_user.id,
            content=payload["content"],
        )


# #endregion export_validation


# #region get_clarification_state [C:3] [TYPE Function]
# @BRIEF Return the current clarification session summary and active question payload.
@router.get(
    "/sessions/{session_id}/clarification",
    response_model=ClarificationStateResponse,
    dependencies=[
        Depends(_require_auto_review_flag),
        Depends(_require_clarification_flag),
        Depends(has_permission("dataset:session", "READ")),
    ],
)
async def get_clarification_state(
    session_id: str,
    repository=Depends(_get_repository),
    clarification_engine=Depends(_get_clarification_engine),
    current_user: User = Depends(get_current_user),
):
    with belief_scope("get_clarification_state"):
        session = _get_owned_session_or_404(repository, session_id, current_user)
        if not session.clarification_sessions:
            return _serialize_empty_clarification_state()
        cs = _get_latest_clarification_session_or_404(session)
        question = clarification_engine.build_question_payload(session)
        return _serialize_clarification_state(
            ClarificationStateResult(
                clarification_session=cs,
                current_question=question,
                session=session,
                changed_findings=[],
            )
        )


# #endregion get_clarification_state


# #region resume_clarification [C:3] [TYPE Function]
# @BRIEF Resume clarification mode on the highest-priority unresolved question.
@router.post(
    "/sessions/{session_id}/clarification/resume",
    response_model=ClarificationStateResponse,
    dependencies=[
        Depends(_require_auto_review_flag),
        Depends(_require_clarification_flag),
        Depends(has_permission("dataset:session", "MANAGE")),
    ],
)
async def resume_clarification(
    session_id: str,
    session_version: int = Depends(_require_session_version_header),
    repository=Depends(_get_repository),
    clarification_engine=Depends(_get_clarification_engine),
    current_user: User = Depends(get_current_user),
):
    with belief_scope("resume_clarification"):
        session = _prepare_owned_session_mutation(
            repository, session_id, current_user, session_version
        )
        cs = _get_latest_clarification_session_or_404(session)
        question = clarification_engine.build_question_payload(session)
        return _serialize_clarification_state(
            ClarificationStateResult(
                clarification_session=cs,
                current_question=question,
                session=session,
                changed_findings=[],
            )
        )


# #endregion resume_clarification


# #region record_clarification_answer [C:3] [TYPE Function]
# @BRIEF Persist one clarification answer before advancing the active pointer.
@router.post(
    "/sessions/{session_id}/clarification/answers",
    response_model=ClarificationAnswerResultResponse,
    dependencies=[
        Depends(_require_auto_review_flag),
        Depends(_require_clarification_flag),
        Depends(has_permission("dataset:session", "MANAGE")),
    ],
)
async def record_clarification_answer(
    session_id: str,
    request: ClarificationAnswerRequest,
    session_version: int = Depends(_require_session_version_header),
    repository=Depends(_get_repository),
    clarification_engine=Depends(_get_clarification_engine),
    current_user: User = Depends(get_current_user),
):
    with belief_scope("dataset_review.record_clarification_answer"):
        session = _prepare_owned_session_mutation(
            repository, session_id, current_user, session_version
        )
        try:
            result = clarification_engine.record_answer(
                ClarificationAnswerCommand(
                    session=session,
                    question_id=request.question_id,
                    answer_kind=request.answer_kind,
                    answer_value=request.answer_value,
                    user=current_user,
                )
            )
        except ValueError as exc:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)
            ) from exc
        return ClarificationAnswerResultResponse(
            clarification_state=_serialize_clarification_state(result),
            session=_serialize_session_summary(result.session),
            changed_findings=[
                ValidationFindingDto.model_validate(f, from_attributes=True)
                for f in result.changed_findings
            ],
        )


# #endregion record_clarification_answer


# #region update_field_semantic [C:3] [TYPE Function]
# @BRIEF Apply one field-level semantic candidate decision or manual override.
@router.patch(
    "/sessions/{session_id}/fields/{field_id}/semantic",
    response_model=SemanticFieldEntryDto,
    dependencies=[
        Depends(_require_auto_review_flag),
        Depends(has_permission("dataset:session", "MANAGE")),
    ],
)
async def update_field_semantic(
    session_id: str,
    field_id: str,
    request: FieldSemanticUpdateRequest,
    session_version: int = Depends(_require_session_version_header),
    repository=Depends(_get_repository),
    current_user: User = Depends(get_current_user),
):
    with belief_scope("dataset_review.update_field_semantic"):
        session = _prepare_owned_session_mutation(
            repository, session_id, current_user, session_version
        )
        field = _get_owned_field_or_404(session, field_id)
        _update_semantic_field_state(field, request, changed_by="user")
        sr = cast(Any, session)
        _commit_owned_session_mutation(repository, session, refresh_targets=[field])
        _record_session_event(
            repository,
            session,
            current_user,
            event_type="semantic_field_updated",
            event_summary="Semantic field decision persisted",
            event_details={"field_id": field.field_id, "version": sr.version},
        )
        return _serialize_semantic_field(field)


# #endregion update_field_semantic


# #region lock_field_semantic [C:3] [TYPE Function]
# @BRIEF Lock one semantic field against later automatic overwrite.
@router.post(
    "/sessions/{session_id}/fields/{field_id}/lock",
    response_model=SemanticFieldEntryDto,
    dependencies=[
        Depends(_require_auto_review_flag),
        Depends(has_permission("dataset:session", "MANAGE")),
    ],
)
async def lock_field_semantic(
    session_id: str,
    field_id: str,
    session_version: int = Depends(_require_session_version_header),
    repository=Depends(_get_repository),
    current_user: User = Depends(get_current_user),
):
    with belief_scope("dataset_review.lock_field_semantic"):
        session = _prepare_owned_session_mutation(
            repository, session_id, current_user, session_version
        )
        field = _get_owned_field_or_404(session, field_id)
        field.is_locked = True
        field.last_changed_by = "user"
        sr = cast(Any, session)
        _commit_owned_session_mutation(repository, session, refresh_targets=[field])
        _record_session_event(
            repository,
            session,
            current_user,
            event_type="semantic_field_locked",
            event_summary="Semantic field lock persisted",
            event_details={"field_id": field.field_id, "version": sr.version},
        )
        return _serialize_semantic_field(field)


# #endregion lock_field_semantic


# #region unlock_field_semantic [C:3] [TYPE Function]
# @BRIEF Unlock one semantic field so later automated candidate application may replace it.
@router.post(
    "/sessions/{session_id}/fields/{field_id}/unlock",
    response_model=SemanticFieldEntryDto,
    dependencies=[
        Depends(_require_auto_review_flag),
        Depends(has_permission("dataset:session", "MANAGE")),
    ],
)
async def unlock_field_semantic(
    session_id: str,
    field_id: str,
    session_version: int = Depends(_require_session_version_header),
    repository=Depends(_get_repository),
    current_user: User = Depends(get_current_user),
):
    with belief_scope("dataset_review.unlock_field_semantic"):
        session = _prepare_owned_session_mutation(
            repository, session_id, current_user, session_version
        )
        field = _get_owned_field_or_404(session, field_id)
        field.is_locked = False
        field.last_changed_by = "user"
        if field.provenance == FieldProvenance.MANUAL_OVERRIDE:
            field.provenance = FieldProvenance.UNRESOLVED
            field.needs_review = True
        sr = cast(Any, session)
        _commit_owned_session_mutation(repository, session, refresh_targets=[field])
        _record_session_event(
            repository,
            session,
            current_user,
            event_type="semantic_field_unlocked",
            event_summary="Semantic field unlock persisted",
            event_details={"field_id": field.field_id, "version": sr.version},
        )
        return _serialize_semantic_field(field)


# #endregion unlock_field_semantic


# #region approve_batch_semantic_fields [C:3] [TYPE Function]
# @BRIEF Approve multiple semantic candidate decisions in one batch.
@router.post(
    "/sessions/{session_id}/fields/semantic/approve-batch",
    response_model=list[SemanticFieldEntryDto],
    dependencies=[
        Depends(_require_auto_review_flag),
        Depends(has_permission("dataset:session", "MANAGE")),
    ],
)
async def approve_batch_semantic_fields(
    session_id: str,
    request: BatchApproveSemanticRequest,
    session_version: int = Depends(_require_session_version_header),
    repository=Depends(_get_repository),
    current_user: User = Depends(get_current_user),
):
    with belief_scope("dataset_review.approve_batch_semantic_fields"):
        session = _prepare_owned_session_mutation(
            repository, session_id, current_user, session_version
        )
        updated = []
        for item in request.items:
            field = _get_owned_field_or_404(session, item.field_id)
            _update_semantic_field_state(
                field,
                FieldSemanticUpdateRequest(
                    candidate_id=item.candidate_id, lock_field=item.lock_field
                ),
                changed_by="user",
            )
            updated.append(field)
        sr = cast(Any, session)
        _commit_owned_session_mutation(
            repository, session, refresh_targets=list(updated)
        )
        _record_session_event(
            repository,
            session,
            current_user,
            event_type="semantic_fields_batch_approved",
            event_summary="Batch semantic approval persisted",
            event_details={"count": len(updated), "version": sr.version},
        )
        return [_serialize_semantic_field(f) for f in updated]


# #endregion approve_batch_semantic_fields


# #region list_execution_mappings [C:3] [TYPE Function]
# @BRIEF Return the current mapping-review set for one accessible session.
@router.get(
    "/sessions/{session_id}/mappings",
    response_model=MappingCollectionResponse,
    dependencies=[
        Depends(_require_auto_review_flag),
        Depends(_require_execution_flag),
        Depends(has_permission("dataset:session", "READ")),
    ],
)
async def list_execution_mappings(
    session_id: str,
    repository=Depends(_get_repository),
    current_user: User = Depends(get_current_user),
):
    with belief_scope("dataset_review.list_execution_mappings"):
        session = _get_owned_session_or_404(repository, session_id, current_user)
        return MappingCollectionResponse(
            items=[_serialize_execution_mapping(m) for m in session.execution_mappings]
        )


# #endregion list_execution_mappings


# #region update_execution_mapping [C:3] [TYPE Function]
# @BRIEF Persist one owner-authorized execution-mapping effective value override.
@router.patch(
    "/sessions/{session_id}/mappings/{mapping_id}",
    response_model=ExecutionMappingDto,
    dependencies=[
        Depends(_require_auto_review_flag),
        Depends(_require_execution_flag),
        Depends(has_permission("dataset:session", "MANAGE")),
    ],
)
async def update_execution_mapping(
    session_id: str,
    mapping_id: str,
    request: UpdateExecutionMappingRequest,
    session_version: int = Depends(_require_session_version_header),
    repository=Depends(_get_repository),
    current_user: User = Depends(get_current_user),
):
    with belief_scope("dataset_review.update_execution_mapping"):
        session = _prepare_owned_session_mutation(
            repository, session_id, current_user, session_version
        )
        mapping = _get_owned_mapping_or_404(session, mapping_id)
        if request.effective_value is None:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="effective_value is required",
            )
        mapping.effective_value = request.effective_value
        mapping.mapping_method = MappingMethod(
            request.mapping_method or MappingMethod.MANUAL_OVERRIDE.value
        )
        mapping.transformation_note = request.transformation_note
        mapping.approval_state = ApprovalState.APPROVED
        mapping.approved_by_user_id = current_user.id
        mapping.approved_at = datetime.now(UTC)
        session.last_activity_at = datetime.now(UTC)
        session.recommended_action = RecommendedAction.GENERATE_SQL_PREVIEW
        if session.readiness_state in {
            ReadinessState.MAPPING_REVIEW_NEEDED,
            ReadinessState.COMPILED_PREVIEW_READY,
            ReadinessState.RUN_READY,
            ReadinessState.RUN_IN_PROGRESS,
        }:
            session.readiness_state = ReadinessState.COMPILED_PREVIEW_READY
        for preview in session.previews:
            if preview.preview_status == PreviewStatus.READY:
                preview.preview_status = PreviewStatus.STALE
        sr = cast(Any, session)
        _commit_owned_session_mutation(repository, session, refresh_targets=[mapping])
        _record_session_event(
            repository,
            session,
            current_user,
            event_type="execution_mapping_updated",
            event_summary="Mapping override persisted",
            event_details={"mapping_id": mapping.mapping_id, "version": sr.version},
        )
        return _serialize_execution_mapping(mapping)


# #endregion update_execution_mapping


# #region approve_execution_mapping [C:3] [TYPE Function]
# @BRIEF Explicitly approve a warning-sensitive mapping transformation.
@router.post(
    "/sessions/{session_id}/mappings/{mapping_id}/approve",
    response_model=ExecutionMappingDto,
    dependencies=[
        Depends(_require_auto_review_flag),
        Depends(_require_execution_flag),
        Depends(has_permission("dataset:session", "MANAGE")),
    ],
)
async def approve_execution_mapping(
    session_id: str,
    mapping_id: str,
    request: ApproveMappingRequest,
    session_version: int = Depends(_require_session_version_header),
    repository=Depends(_get_repository),
    current_user: User = Depends(get_current_user),
):
    with belief_scope("dataset_review.approve_execution_mapping"):
        session = _prepare_owned_session_mutation(
            repository, session_id, current_user, session_version
        )
        mapping = _get_owned_mapping_or_404(session, mapping_id)
        mapping.approval_state = ApprovalState.APPROVED
        mapping.approved_by_user_id = current_user.id
        mapping.approved_at = datetime.now(UTC)
        if request.approval_note:
            mapping.transformation_note = request.approval_note
        session.last_activity_at = datetime.now(UTC)
        if session.readiness_state == ReadinessState.MAPPING_REVIEW_NEEDED:
            session.recommended_action = RecommendedAction.GENERATE_SQL_PREVIEW
        sr = cast(Any, session)
        _commit_owned_session_mutation(repository, session, refresh_targets=[mapping])
        _record_session_event(
            repository,
            session,
            current_user,
            event_type="execution_mapping_approved",
            event_summary="Mapping approval persisted",
            event_details={"mapping_id": mapping.mapping_id, "version": sr.version},
        )
        return _serialize_execution_mapping(mapping)


# #endregion approve_execution_mapping


# #region approve_batch_execution_mappings [C:3] [TYPE Function]
# @BRIEF Approve multiple warning-sensitive execution mappings in one batch.
@router.post(
    "/sessions/{session_id}/mappings/approve-batch",
    response_model=list[ExecutionMappingDto],
    dependencies=[
        Depends(_require_auto_review_flag),
        Depends(_require_execution_flag),
        Depends(has_permission("dataset:session", "MANAGE")),
    ],
)
async def approve_batch_execution_mappings(
    session_id: str,
    request: BatchApproveMappingRequest,
    session_version: int = Depends(_require_session_version_header),
    repository=Depends(_get_repository),
    current_user: User = Depends(get_current_user),
):
    with belief_scope("dataset_review.approve_batch_execution_mappings"):
        session = _prepare_owned_session_mutation(
            repository, session_id, current_user, session_version
        )
        updated = []
        for mid in list(dict.fromkeys(request.mapping_ids)):
            mapping = _get_owned_mapping_or_404(session, mid)
            mapping.approval_state = ApprovalState.APPROVED
            mapping.approved_by_user_id = current_user.id
            mapping.approved_at = datetime.now(UTC)
            if request.approval_note:
                mapping.transformation_note = request.approval_note
            updated.append(mapping)
        session.last_activity_at = datetime.now(UTC)
        if session.readiness_state == ReadinessState.MAPPING_REVIEW_NEEDED:
            session.recommended_action = RecommendedAction.GENERATE_SQL_PREVIEW
        sr = cast(Any, session)
        _commit_owned_session_mutation(
            repository, session, refresh_targets=list(updated)
        )
        _record_session_event(
            repository,
            session,
            current_user,
            event_type="execution_mappings_batch_approved",
            event_summary="Batch mapping approval persisted",
            event_details={"count": len(updated), "version": sr.version},
        )
        return [_serialize_execution_mapping(m) for m in updated]


# #endregion approve_batch_execution_mappings


# #region trigger_preview_generation [C:3] [TYPE Function]
# @BRIEF Trigger Superset-side preview compilation for the current owned execution context.
@router.post(
    "/sessions/{session_id}/preview",
    response_model=Union[CompiledPreviewDto, PreviewEnqueueResultResponse],
    dependencies=[
        Depends(_require_auto_review_flag),
        Depends(_require_execution_flag),
        Depends(has_permission("dataset:session", "MANAGE")),
    ],
)
async def trigger_preview_generation(
    session_id: str,
    response: Response,
    orchestrator=Depends(_get_orchestrator),
    repository=Depends(_get_repository),
    session_version: int = Depends(_require_session_version_header),
    current_user: User = Depends(get_current_user),
):
    with belief_scope("dataset_review.trigger_preview_generation"):
        _prepare_owned_session_mutation(
            repository, session_id, current_user, session_version
        )
        try:
            result = orchestrator.prepare_launch_preview(
                PreparePreviewCommand(
                    user=current_user,
                    session_id=session_id,
                    expected_version=session_version,
                )
            )
        except DatasetReviewSessionVersionConflictError as exc:
            raise _build_session_version_conflict_http_exception(exc) from exc
        except ValueError as exc:
            detail = str(exc)
            sc = (
                status.HTTP_404_NOT_FOUND
                if detail in {"Session not found", "Environment not found"}
                else status.HTTP_409_CONFLICT
                if detail.startswith("Preview blocked:")
                else status.HTTP_400_BAD_REQUEST
            )
            raise HTTPException(status_code=sc, detail=detail) from exc
        if result.preview.preview_status == PreviewStatus.PENDING:
            response.status_code = status.HTTP_202_ACCEPTED
            return PreviewEnqueueResultResponse(
                session_id=result.session.session_id,
                session_version=int(getattr(result.session, "version", 0) or 0),
                preview_status=result.preview.preview_status.value,
                task_id=None,
            )
        response.status_code = status.HTTP_200_OK
        return _serialize_preview(
            result.preview,
            session_version_fallback=int(getattr(result.session, "version", 0) or 0),
        )


# #endregion trigger_preview_generation


# #region launch_dataset [C:3] [TYPE Function]
# @BRIEF Execute the current owned session launch handoff and return audited SQL Lab run context.
@router.post(
    "/sessions/{session_id}/launch",
    response_model=LaunchDatasetResponse,
    status_code=status.HTTP_201_CREATED,
    dependencies=[
        Depends(_require_auto_review_flag),
        Depends(_require_execution_flag),
        Depends(has_permission("dataset:execution:launch", "EXECUTE")),
    ],
)
async def launch_dataset(
    session_id: str,
    orchestrator=Depends(_get_orchestrator),
    repository=Depends(_get_repository),
    session_version: int = Depends(_require_session_version_header),
    config_manager=Depends(get_config_manager),
    current_user: User = Depends(get_current_user),
):
    with belief_scope("dataset_review.launch_dataset"):
        _prepare_owned_session_mutation(
            repository, session_id, current_user, session_version
        )
        try:
            result = orchestrator.launch_dataset(
                LaunchDatasetCommand(
                    user=current_user,
                    session_id=session_id,
                    expected_version=session_version,
                )
            )
        except DatasetReviewSessionVersionConflictError as exc:
            raise _build_session_version_conflict_http_exception(exc) from exc
        except ValueError as exc:
            detail = str(exc)
            sc = (
                status.HTTP_404_NOT_FOUND
                if detail in {"Session not found", "Environment not found"}
                else status.HTTP_409_CONFLICT
                if detail.startswith("Launch blocked:")
                else status.HTTP_400_BAD_REQUEST
            )
            raise HTTPException(status_code=sc, detail=detail) from exc
        environment = config_manager.get_environment(result.session.environment_id)
        env_url = getattr(environment, "url", "") if environment is not None else ""
        return LaunchDatasetResponse(
            run_context=_serialize_run_context(result.run_context),
            redirect_url=_build_sql_lab_redirect_url(
                environment_url=env_url,
                sql_lab_session_ref=result.run_context.sql_lab_session_ref,
            ),
        )


# #endregion launch_dataset


# #region record_field_feedback [C:3] [TYPE Function]
# @BRIEF Persist thumbs up/down feedback for AI-assisted semantic field content.
@router.post(
    "/sessions/{session_id}/fields/{field_id}/feedback",
    response_model=FeedbackResponse,
    dependencies=[
        Depends(_require_auto_review_flag),
        Depends(has_permission("dataset:session", "MANAGE")),
    ],
)
async def record_field_feedback(
    session_id: str,
    field_id: str,
    request: FeedbackRequest,
    session_version: int = Depends(_require_session_version_header),
    repository=Depends(_get_repository),
    current_user: User = Depends(get_current_user),
):
    with belief_scope("dataset_review.record_field_feedback"):
        session = _prepare_owned_session_mutation(
            repository, session_id, current_user, session_version
        )
        field = _get_owned_field_or_404(session, field_id)
        field.user_feedback = request.feedback
        sr = cast(Any, session)
        _commit_owned_session_mutation(repository, session)
        _record_session_event(
            repository,
            session,
            current_user,
            event_type="semantic_field_feedback_recorded",
            event_summary="Feedback persisted",
            event_details={
                "field_id": field.field_id,
                "feedback": request.feedback,
                "version": sr.version,
            },
        )
        return FeedbackResponse(target_id=field.field_id, feedback=request.feedback)


# #endregion record_field_feedback


# #region record_clarification_feedback [C:3] [TYPE Function]
# @BRIEF Persist thumbs up/down feedback for clarification question/answer content.
@router.post(
    "/sessions/{session_id}/clarification/questions/{question_id}/feedback",
    response_model=FeedbackResponse,
    dependencies=[
        Depends(_require_auto_review_flag),
        Depends(_require_clarification_flag),
        Depends(has_permission("dataset:session", "MANAGE")),
    ],
)
async def record_clarification_feedback(
    session_id: str,
    question_id: str,
    request: FeedbackRequest,
    session_version: int = Depends(_require_session_version_header),
    repository=Depends(_get_repository),
    current_user: User = Depends(get_current_user),
):
    with belief_scope("dataset_review.record_clarification_feedback"):
        session = _prepare_owned_session_mutation(
            repository, session_id, current_user, session_version
        )
        cs = _get_latest_clarification_session_or_404(session)
        question = next((q for q in cs.questions if q.question_id == question_id), None)
        if question is None:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail="Clarification question not found",
            )
        if question.answer is None:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="Clarification answer not found",
            )
        question.answer.user_feedback = request.feedback
        sr = cast(Any, session)
        _commit_owned_session_mutation(repository, session)
        _record_session_event(
            repository,
            session,
            current_user,
            event_type="clarification_feedback_recorded",
            event_summary="Feedback persisted",
            event_details={
                "question_id": question.question_id,
                "feedback": request.feedback,
                "version": sr.version,
            },
        )
        return FeedbackResponse(
            target_id=question.question_id, feedback=request.feedback
        )


# #endregion record_clarification_feedback


# #endregion DatasetReviewRoutes
