"""Durable runtime for Campaign intelligence jobs.

智能分析与周期对比是两个领域工作 adapter；本 module 的 ``execute``
统一掌握它们的结算契约：建行/认领/校验/generating/failed/completed。
数据库行是耐久权威，TaskRegistry 只保存当前进程中的任务句柄。
"""

import logging
from collections.abc import Awaitable, Callable
from typing import Any, Optional

from sqlmodel import Session

from agenteval.storage.async_job_repository import CampaignAnalysisRepository, CampaignPeriodComparisonRepository
from agenteval.storage.db import get_session
from agenteval.task_registry import TaskRegistry

_registry = TaskRegistry()
_logger = logging.getLogger("agenteval")
MAX_QUEUED_RECOVERY_ATTEMPTS = 3
_ERROR_TRUNCATE_LEN = 500

# adapter 契约：
# ensure_queued(session) -> row | None（建行或确认已有行；None = 静默放弃）
# validate(session) -> 落账附加字段（认领后校验，失败抛 JobValidationError）
# work_fn(session) -> (result, 落账附加字段)
EnsureQueuedFn = Callable[[Session], Any]
ValidateFn = Callable[[Session], Optional[dict]]
WorkFn = Callable[[Session], Awaitable[tuple[dict, dict]]]


class JobValidationError(RuntimeError):
    """认领后校验失败：携带已确定的落账字段（如对比的生效基线）进 failed 行。"""

    def __init__(self, message: str, **meta: Any):
        super().__init__(message)
        self.meta = meta


def _job_key(kind: str, campaign_id: str) -> str:
    return f"{kind}:{campaign_id}"


def _launch_analysis(campaign_id: str, *, triggered_by: str) -> None:
    async def run(_cancel) -> None:
        await execute_campaign_analysis_job(campaign_id, triggered_by=triggered_by)

    _registry.launch(_job_key("analysis", campaign_id), run)


def _launch_comparison(
    campaign_id: str,
    *,
    triggered_by: str,
    baseline_campaign_id: str,
) -> None:
    async def run(_cancel) -> None:
        await execute_campaign_comparison_job(
            campaign_id,
            triggered_by=triggered_by,
            baseline_campaign_id=baseline_campaign_id,
        )

    _registry.launch(_job_key("comparison", campaign_id), run)


async def execute(
    job_kind: str,
    campaign_id: str,
    *,
    triggered_by: str,
    repo_cls,
    ensure_queued: EnsureQueuedFn,
    validate: Optional[ValidateFn] = None,
    work_fn: WorkFn,
    session_factory: Optional[Callable[[], Session]] = None,
) -> None:
    """结算一个耐久智能作业：建行 → 认领 → 校验 → generating → 落账。

    结算契约（两个 adapter 共用，只此一处）：

    - ``ensure_queued`` 幂等建行（或确认已有行）；返回 None 表示静默
      放弃（如活动不存在），不写任何行。
    - ``claim_queued`` 是耐久幂等权威：抢不到（重复触发 / 已在执行）
      静默返回，绝不重复跑领域工作。
    - ``validate`` 在认领后校验并产出附加字段（模型、基线），随
      generating/结算落账；抛 ``JobValidationError`` → 落 failed 行
      （异常携带的已确定字段一并落账）。
    - ``work_fn`` 抛任何异常 → failed 行，error 截断到 500 字符（一处）。
    """
    session = (session_factory or get_session)()
    repo = repo_cls(session)
    try:
        row = ensure_queued(session)
        if row is None:
            return
        if not repo.claim_queued(campaign_id).claimed:
            return
        effective_trigger = row.triggered_by or triggered_by

        meta: dict[str, Any] = {}
        if validate is not None:
            try:
                meta = validate(session) or {}
            except JobValidationError as exc:
                repo.upsert(
                    campaign_id,
                    status="failed",
                    triggered_by=effective_trigger,
                    error=str(exc),
                    **exc.meta,
                )
                return

        repo.upsert(campaign_id, status="generating", triggered_by=effective_trigger, **meta)
        work_meta: dict[str, Any] = {}
        try:
            result, work_meta = await work_fn(session)
        except Exception as exc:
            _logger.warning("%s 作业 %s 失败: %s", job_kind, campaign_id, exc)
            repo.upsert(
                campaign_id,
                status="failed",
                error=str(exc)[:_ERROR_TRUNCATE_LEN],
                triggered_by=effective_trigger,
                **meta,
                **work_meta,
            )
            return
        repo.upsert(
            campaign_id,
            status="completed",
            result=result,
            triggered_by=effective_trigger,
            **meta,
            **work_meta,
        )
    finally:
        session.close()


async def execute_campaign_analysis_job(
    campaign_id: str,
    *,
    triggered_by: str,
    chat_client: Any = None,
    session_factory: Optional[Callable[[], Session]] = None,
) -> None:
    """Claim and settle one intelligent-analysis job (work adapter)."""

    def ensure_queued(session: Session):
        analyses = CampaignAnalysisRepository(session)
        row = analyses.get_by_campaign(campaign_id)
        return row or analyses.enqueue(campaign_id, triggered_by=triggered_by)

    def validate(session: Session) -> dict:
        from agenteval.evaluation.analysis import AnalysisError, validate_analysis_request
        from agenteval.storage.repository import CampaignRepository

        campaign = CampaignRepository(session).get(campaign_id)
        if campaign is None:
            raise JobValidationError("campaign not found")
        try:
            runtime = validate_analysis_request(session, campaign)
        except AnalysisError as exc:
            raise JobValidationError(str(exc)) from exc
        return {"model_config_id": runtime.id}

    async def work(session: Session) -> tuple[dict, dict]:
        from agenteval.evaluation.analysis import (
            analyze_campaign,
            collect_failure_samples,
            gateway_chat_client,
            resolve_analysis_model,
        )
        from agenteval.evaluation.report import load_campaign_view
        from agenteval.storage.repository import CampaignRepository, RunRepository

        campaign = CampaignRepository(session).get(campaign_id)
        runtime = resolve_analysis_model(campaign, session)
        runs = RunRepository(session).list_by_campaign(campaign_id)
        view = load_campaign_view(session, campaign)
        result = await analyze_campaign(
            campaign=campaign,
            report=view["report"],
            failure_samples=collect_failure_samples(campaign_id, session),
            valid_run_ids={run.id for run in runs if run.id},
            chat_client=chat_client or gateway_chat_client(runtime),
            exploration_summary=view["exploration"],
        )
        return result, {}

    await execute(
        "analysis",
        campaign_id,
        triggered_by=triggered_by,
        repo_cls=CampaignAnalysisRepository,
        ensure_queued=ensure_queued,
        validate=validate,
        work_fn=work,
        session_factory=session_factory,
    )
    await _maybe_enqueue_auto_comparison(campaign_id, session_factory=session_factory)


def auto_intelligence_eligible(campaign, session: Session) -> bool:
    """自动触发智能分析/周期对比的唯一跳过守卫：正式线 + 分析模型可解析。

    活动完成时的自动分析与分析完成后的自动对比共用这一判断点；
    加速线与未配置模型一律静默跳过（增强能力，不阻断主流程）。
    """
    from agenteval.evaluation.analysis import resolve_analysis_model

    return bool(campaign.id) and campaign.time_scale == 1 and resolve_analysis_model(campaign, session) is not None


async def _maybe_enqueue_auto_comparison(
    campaign_id: str,
    *,
    session_factory: Optional[Callable[[], Session]] = None,
) -> None:
    """分析完成后：存在自动基线 → 入队周期对比。任何一步不满足或出错都静默跳过。"""
    from agenteval.evaluation.comparison import resolve_auto_baseline
    from agenteval.storage.repository import CampaignRepository

    session = (session_factory or get_session)()
    try:
        campaign = CampaignRepository(session).get(campaign_id)
        if campaign is None or not auto_intelligence_eligible(campaign, session):
            return
        row = CampaignAnalysisRepository(session).get_by_campaign(campaign_id)
        if row is None or row.status != "completed":
            return
        baseline = resolve_auto_baseline(campaign, session)
        if baseline is None:
            return
        enqueue_campaign_comparison(
            campaign.id,
            triggered_by="auto",
            baseline_campaign_id=baseline.id,
            session=session,
        )
    except Exception as exc:
        _logger.warning("活动 %s 自动周期对比跳过: %s", campaign_id, exc)
    finally:
        session.close()


async def execute_campaign_comparison_job(
    campaign_id: str,
    *,
    triggered_by: str,
    baseline_campaign_id: Optional[str] = None,
    chat_client: Any = None,
    session_factory: Optional[Callable[[], Session]] = None,
) -> None:
    """Claim and settle one period-comparison job (work adapter)."""
    from agenteval.evaluation.comparison import ComparisonError

    def ensure_queued(session: Session):
        from agenteval.evaluation.comparison import validate_comparison_request
        from agenteval.storage.repository import CampaignRepository

        comparisons = CampaignPeriodComparisonRepository(session)
        campaign = CampaignRepository(session).get(campaign_id)
        if campaign is None:
            return None
        row = comparisons.get_by_campaign(campaign_id)
        if row is not None:
            return row
        try:
            initial_baseline = validate_comparison_request(
                session,
                campaign,
                explicit_baseline_id=baseline_campaign_id,
            )
        except ComparisonError as exc:
            comparisons.upsert(
                campaign_id,
                status="failed",
                triggered_by=triggered_by,
                error=str(exc),
            )
            return None
        return comparisons.enqueue(
            campaign_id,
            baseline_campaign_id=initial_baseline.id,
            triggered_by=triggered_by,
        )

    def validate(session: Session) -> dict:
        from agenteval.evaluation.analysis import resolve_analysis_model
        from agenteval.evaluation.comparison import (
            ComparisonError,
            validate_comparison_request,
        )
        from agenteval.storage.repository import CampaignRepository

        campaign = CampaignRepository(session).get(campaign_id)
        row = CampaignPeriodComparisonRepository(session).get_by_campaign(campaign_id)
        effective_baseline_id = baseline_campaign_id or (row.baseline_campaign_id if row else None)
        try:
            baseline = validate_comparison_request(
                session,
                campaign,
                explicit_baseline_id=effective_baseline_id,
            )
        except ComparisonError as exc:
            raise JobValidationError(str(exc), baseline_campaign_id=effective_baseline_id) from exc
        runtime = resolve_analysis_model(campaign, session)
        return {"baseline_campaign_id": baseline.id, "model_config_id": runtime.id}

    async def work(session: Session) -> tuple[dict, dict]:
        from agenteval.evaluation.analysis import gateway_chat_client, resolve_analysis_model
        from agenteval.evaluation.comparison import compute_metric_diff, narrate_period_comparison
        from agenteval.evaluation.report import load_campaign_report
        from agenteval.storage.repository import CampaignRepository

        campaign = CampaignRepository(session).get(campaign_id)
        row = CampaignPeriodComparisonRepository(session).get_by_campaign(campaign_id)
        effective_baseline_id = baseline_campaign_id or (row.baseline_campaign_id if row else None)
        baseline = CampaignRepository(session).get(effective_baseline_id)
        runtime = resolve_analysis_model(campaign, session)
        baseline_analysis = CampaignAnalysisRepository(session).get_by_campaign(baseline.id)
        current_analysis = CampaignAnalysisRepository(session).get_by_campaign(campaign_id)
        diff = compute_metric_diff(
            load_campaign_report(session, baseline),
            load_campaign_report(session, campaign),
        )
        result = await narrate_period_comparison(
            baseline_analysis=baseline_analysis.get_result(),
            current_analysis=current_analysis.get_result(),
            metric_diff=diff,
            valid_scenario_ids={item["scenario_id"] for item in diff["scenarios"]},
            chat_client=chat_client or gateway_chat_client(runtime),
        )
        return result, {}

    await execute(
        "comparison",
        campaign_id,
        triggered_by=triggered_by,
        repo_cls=CampaignPeriodComparisonRepository,
        ensure_queued=ensure_queued,
        validate=validate,
        work_fn=work,
        session_factory=session_factory,
    )


def enqueue_campaign_analysis(
    campaign_id: str,
    *,
    triggered_by: str,
    session: Optional[Session] = None,
) -> None:
    """Persist an analysis job, then launch its process-local worker."""
    owns_session = session is None
    active_session = session or get_session()
    try:
        row = CampaignAnalysisRepository(active_session).enqueue(
            campaign_id,
            triggered_by=triggered_by,
        )
    finally:
        if owns_session:
            active_session.close()
    if row.status == "queued":
        _launch_analysis(campaign_id, triggered_by=row.triggered_by or triggered_by)


def enqueue_campaign_comparison(
    campaign_id: str,
    *,
    triggered_by: str,
    baseline_campaign_id: str,
    session: Optional[Session] = None,
) -> None:
    """Persist a comparison job, then launch its process-local worker."""
    owns_session = session is None
    active_session = session or get_session()
    try:
        row = CampaignPeriodComparisonRepository(active_session).enqueue(
            campaign_id,
            baseline_campaign_id=baseline_campaign_id,
            triggered_by=triggered_by,
        )
    finally:
        if owns_session:
            active_session.close()
    if row.status == "queued":
        _launch_comparison(
            campaign_id,
            triggered_by=row.triggered_by or triggered_by,
            baseline_campaign_id=row.baseline_campaign_id,
        )


def recover_campaign_intelligence_jobs(session: Session) -> tuple[int, int]:
    """Fail interrupted work and relaunch every durably queued job."""
    analyses = CampaignAnalysisRepository(session)
    comparisons = CampaignPeriodComparisonRepository(session)
    interrupted = analyses.mark_orphans_failed() + comparisons.mark_orphans_failed()

    queued_analyses = analyses.prepare_queued_recovery(
        MAX_QUEUED_RECOVERY_ATTEMPTS,
        "服务重启恢复次数超过上限，分析任务已终止",
    )
    for row in queued_analyses:
        _launch_analysis(row.campaign_id, triggered_by=row.triggered_by or "manual")

    queued_comparisons = comparisons.prepare_queued_recovery(
        MAX_QUEUED_RECOVERY_ATTEMPTS,
        "服务重启恢复次数超过上限，周期对比任务已终止",
    )
    for row in queued_comparisons:
        _launch_comparison(
            row.campaign_id,
            triggered_by=row.triggered_by or "manual",
            baseline_campaign_id=row.baseline_campaign_id,
        )
    return interrupted, len(queued_analyses) + len(queued_comparisons)


def is_intelligence_job_running(kind: str, campaign_id: str) -> bool:
    """Expose process-local liveness without exposing registry internals."""
    return _registry.is_running(_job_key(kind, campaign_id))


async def shutdown_campaign_intelligence_jobs() -> None:
    """Stop every live analysis and comparison worker."""
    await _registry.shutdown_all()