"""Run 创建、查询与内置模拟执行器。"""
from __future__ import annotations

import asyncio
import hashlib
import json
from typing import Any, Dict, Optional, Set

from sqlalchemy import delete, select
from sqlalchemy.ext.asyncio import AsyncSession

from ..config import get_settings
from ..db import session_scope
from ..enums import (
    EventKind,
    EventSource,
    HostStatus,
    RecoveryLevel,
    RecoveryOutcome,
    RecoveryTrigger,
    RunState,
    Severity,
    Verdict,
)
from ..events import append_event
from ..evidence import build_evidence_bundle
from ..engine.planner import materialize_run_steps
from ..models import (
    Dut,
    FirmwareArtifact,
    OobController,
    RecoveryAction,
    ResourceLock,
    Run,
    RunEvent,
    TestHost,
    Workflow,
    iso_z,
    utcnow,
)
from ..safety import PreflightRequest, run_preflight
from ..schemas import CreateRunBody, PreflightBody
from .agent_status import refresh_agent_host

_tasks: Dict[str, asyncio.Task[None]] = {}
_task_refs: Set[asyncio.Task[None]] = set()


async def _first(session: AsyncSession, model: Any) -> Any:
    value = await session.scalar(select(model).limit(1))
    if value is None:
        raise LookupError("演示数据尚未初始化")
    return value


async def resolve_assets(
    session: AsyncSession,
    *,
    dut_id: Optional[str] = None,
    host_id: Optional[str] = None,
    firmware_id: Optional[str] = None,
) -> tuple[Dut, TestHost, Optional[FirmwareArtifact], Optional[OobController]]:
    dut = await session.get(Dut, dut_id) if dut_id else await _first(session, Dut)
    host = await session.get(TestHost, host_id) if host_id else await _first(session, TestHost)
    firmware = (
        await session.get(FirmwareArtifact, firmware_id)
        if firmware_id
        else await _first(session, FirmwareArtifact)
    )
    if dut is None or host is None:
        raise LookupError("DUT 或测试主机不存在")
    oob = await session.get(OobController, host.oob_controller_id) if host.oob_controller_id else None
    return dut, host, firmware, oob


async def preflight_for_body(session: AsyncSession, body: PreflightBody) -> Dict[str, object]:
    dut, host, firmware, oob = await resolve_assets(
        session,
        dut_id=body.dut_id,
        host_id=body.host_id,
        firmware_id=body.firmware_id,
    )
    refresh_agent_host(host)
    system_paths = host.labels.get("system_device_paths", []) if host.labels else []
    allowed = firmware is not None and (
        not firmware.applicable_models or dut.model in firmware.applicable_models
    )
    result = run_preflight(
        PreflightRequest(
            expected_serial=dut.serial,
            discovered_serial=body.discovered_serial or dut.serial,
            allow_destructive=dut.allow_destructive,
            device_path=dut.device_path or "",
            system_device_paths=system_paths,
            firmware_model_allowed=allowed,
            host_online=host.status == HostStatus.ONLINE.value,
            oob_online=oob is not None and oob.status == HostStatus.ONLINE.value,
        )
    )
    payload = result.as_dict()
    payload.update({"dut_id": dut.id, "host_id": host.id, "firmware_id": firmware.id if firmware else None})
    return payload


async def create_run(session: AsyncSession, body: CreateRunBody) -> Run:
    workflow = (
        await session.get(Workflow, body.workflow_id)
        if body.workflow_id
        else await _first(session, Workflow)
    )
    dut, host, _, _ = await resolve_assets(
        session, dut_id=body.dut_id, host_id=body.host_id
    )
    firmwares = (
        await session.scalars(select(FirmwareArtifact).order_by(FirmwareArtifact.created_at))
    ).all()
    if not firmwares:
        raise LookupError("固件数据尚未初始化")
    firmware_a = (
        await session.get(FirmwareArtifact, body.firmware_a_id)
        if body.firmware_a_id
        else firmwares[0]
    )
    firmware_b = (
        await session.get(FirmwareArtifact, body.firmware_b_id)
        if body.firmware_b_id
        else firmwares[-1]
    )
    assert workflow is not None and firmware_a is not None and firmware_b is not None
    env = {
        "host": host.name,
        "os": host.os_version,
        "kernel": host.kernel,
        "bios": host.bios_version,
        "agent": host.agent_version,
        "toolchain": host.toolchain,
        "dut_serial": dut.serial,
        "dut_model": dut.model,
    }
    env_hash = hashlib.sha256(
        json.dumps(env, ensure_ascii=False, sort_keys=True).encode("utf-8")
    ).hexdigest()
    run = Run(
        name=body.name,
        workflow_id=workflow.id,
        workflow_key=workflow.key,
        workflow_version=workflow.version,
        spec_hash=workflow.spec_hash,
        dut_id=dut.id,
        test_host_id=host.id,
        firmware_a_id=firmware_a.id,
        firmware_b_id=firmware_b.id,
        params={
            "inject_failure": body.inject_failure,
            "execution_mode": body.execution_mode,
            "loops": body.loops,
            "fw_a": firmware_a.id,
            "fw_b": firmware_b.id,
            "firmware_a_version": firmware_a.version,
            "firmware_b_version": firmware_b.version,
        },
        env_fingerprint=env,
        env_fingerprint_hash=env_hash,
        loop_target=body.loops,
        created_by=body.created_by,
    )
    session.add(run)
    await session.flush()
    await append_event(
        session,
        run,
        EventKind.RUN_CREATED.value,
        "任务已创建，等待独占资源",
        payload={"loops": body.loops, "inject_failure": body.inject_failure},
    )
    await append_event(
        session, run, EventKind.RUN_QUEUED.value, "任务已进入调度队列"
    )
    return run


def run_to_dict(run: Run) -> Dict[str, Any]:
    progress = round((run.loop_done / run.loop_target) * 100, 1) if run.loop_target else 0
    return {
        "id": run.id,
        "name": run.name,
        "state": run.state,
        "verdict": run.verdict,
        "workflow_key": run.workflow_key,
        "workflow_version": run.workflow_version,
        "execution_mode": run.params.get("execution_mode", "simulated"),
        "dut_id": run.dut_id,
        "test_host_id": run.test_host_id,
        "loop_target": run.loop_target,
        "loop_done": run.loop_done,
        "progress": progress,
        "checkpoint": run.checkpoint,
        "environment": run.env_fingerprint,
        "firmware_a_version": run.params.get("firmware_a_version"),
        "firmware_b_version": run.params.get("firmware_b_version"),
        "recovery_count": run.recovery_count,
        "unattended_completion": run.unattended_completion,
        "human_touches": run.human_touches,
        "created_at": iso_z(run.created_at),
        "started_at": iso_z(run.started_at),
        "ended_at": iso_z(run.ended_at),
        "freeze_reason": run.freeze_reason,
    }


async def prepare_agent_run(session: AsyncSession, run: Run) -> None:
    """Run safety gates, lock resources and materialize an Agent dry-run."""

    await append_event(
        session,
        run,
        EventKind.RUN_PREFLIGHT_STARTED.value,
        "已锁定执行计划，开始 Agent 任务安全预检",
        target_state=RunState.PREFLIGHT.value,
    )
    preflight = await preflight_for_body(
        session,
        PreflightBody(
            dut_id=run.dut_id,
            host_id=run.test_host_id,
            firmware_id=run.firmware_b_id,
        ),
    )
    if not preflight["passed"]:
        await append_event(
            session,
            run,
            EventKind.RUN_REJECTED.value,
            "安全预检未通过",
            severity=Severity.CRITICAL.value,
            payload=preflight,
            target_state=RunState.REJECTED.value,
            projection={"ended_at": utcnow()},
        )
        return

    existing_lock = await session.scalar(
        select(ResourceLock.id).where(
            ResourceLock.resource_type.in_(["dut", "test_host"]),
            ResourceLock.resource_id.in_([run.dut_id or "", run.test_host_id or ""]),
        )
    )
    if existing_lock:
        raise LookupError("DUT 或测试主机已被其他任务锁定")

    workflow = await session.get(Workflow, run.workflow_id)
    if workflow is None:
        raise LookupError("Run 关联工作流不存在")
    steps = await materialize_run_steps(session, run, workflow)
    session.add_all(
        [
            ResourceLock(
                resource_type="dut",
                resource_id=run.dut_id or "",
                run_id=run.id,
                acquired_at=utcnow(),
            ),
            ResourceLock(
                resource_type="test_host",
                resource_id=run.test_host_id or "",
                run_id=run.id,
                acquired_at=utcnow(),
            ),
        ]
    )
    await append_event(
        session,
        run,
        EventKind.SAFETY_APPROVED.value,
        "6/6 安全门禁通过",
        payload=preflight,
    )
    await append_event(
        session,
        run,
        EventKind.LOCK_ACQUIRED.value,
        "DUT 与测试主机独占锁已获取",
    )
    await append_event(
        session,
        run,
        EventKind.RUN_STARTED.value,
        "独立 Host Agent dry-run 已就绪，共 {0} 个步骤".format(len(steps)),
        payload={"step_count": len(steps), "execution_mode": "agent_dry_run"},
        target_state=RunState.RUNNING.value,
        projection={"started_at": utcnow()},
    )


def event_to_dict(event: RunEvent) -> Dict[str, Any]:
    return {
        "id": event.id,
        "seq": event.seq,
        "ts": iso_z(event.ts),
        "source": event.source,
        "kind": event.kind,
        "severity": event.severity,
        "message": event.message,
        "payload": event.payload,
    }


async def _sleep(seconds: float) -> None:
    await asyncio.sleep(max(0.01, seconds * get_settings().time_scale))


async def _load_run(session: AsyncSession, run_id: str) -> Run:
    run = await session.get(Run, run_id)
    if run is None:
        raise LookupError("Run 不存在: {0}".format(run_id))
    return run


async def _run_recovery(run_id: str, loop_index: int) -> None:
    levels = [
        (RecoveryLevel.AGENT_SOFT, RecoveryOutcome.FAILED, "Agent 软恢复无响应"),
        (RecoveryLevel.OS_REBOOT, RecoveryOutcome.FAILED, "OS 重启通道超时"),
        (RecoveryLevel.OOB_RESET, RecoveryOutcome.RECOVERED, "带外 Reset 后主机与 DUT 已恢复"),
    ]
    for level, outcome, message in levels:
        async with session_scope() as session:
            run = await _load_run(session, run_id)
            action = RecoveryAction(
                run_id=run.id,
                loop_index=loop_index,
                level=level.value,
                trigger=RecoveryTrigger.HEARTBEAT_LOST.value,
                outcome=outcome.value,
                started_at=utcnow(),
                ended_at=utcnow(),
                detail={"simulated": True, "message": message},
                evidence_uris=["sim://oob/frame-{0}".format(loop_index)],
            )
            session.add(action)
            await append_event(
                session,
                run,
                EventKind.RECOVERY_STARTED.value,
                "{0} 已执行".format(level.value),
                severity=Severity.WARNING.value,
                payload={"level": level.value, "outcome": outcome.value},
                projection={"recovery_count": run.recovery_count + 1},
            )
            if outcome == RecoveryOutcome.RECOVERED:
                await append_event(
                    session,
                    run,
                    EventKind.RECOVERY_SUCCEEDED.value,
                    message,
                    source=EventSource.OOB.value,
                    payload={"level": level.value},
                    target_state=RunState.RUNNING.value,
                )
            else:
                await append_event(
                    session,
                    run,
                    EventKind.RECOVERY_ESCALATED.value,
                    message + "，升级恢复阶梯",
                    severity=Severity.ERROR.value,
                    payload={"level": level.value},
                )
        await _sleep(0.28)


async def simulate_run(run_id: str) -> None:
    try:
        async with session_scope() as session:
            run = await _load_run(session, run_id)
            await append_event(
                session,
                run,
                EventKind.RUN_PREFLIGHT_STARTED.value,
                "已锁定工位，开始安全预检",
                target_state=RunState.PREFLIGHT.value,
            )
        await _sleep(0.25)

        async with session_scope() as session:
            run = await _load_run(session, run_id)
            preflight = await preflight_for_body(
                session,
                PreflightBody(
                    dut_id=run.dut_id,
                    host_id=run.test_host_id,
                    firmware_id=run.firmware_b_id,
                ),
            )
            if not preflight["passed"]:
                await append_event(
                    session,
                    run,
                    EventKind.RUN_REJECTED.value,
                    "安全预检未通过",
                    severity=Severity.CRITICAL.value,
                    payload=preflight,
                    target_state=RunState.REJECTED.value,
                    projection={"ended_at": utcnow()},
                )
                return
            session.add_all(
                [
                    ResourceLock(
                        resource_type="dut",
                        resource_id=run.dut_id or "",
                        run_id=run.id,
                        acquired_at=utcnow(),
                    ),
                    ResourceLock(
                        resource_type="test_host",
                        resource_id=run.test_host_id or "",
                        run_id=run.id,
                        acquired_at=utcnow(),
                    ),
                ]
            )
            await append_event(
                session,
                run,
                EventKind.SAFETY_APPROVED.value,
                "6/6 安全门禁通过",
                payload=preflight,
            )
            await append_event(
                session,
                run,
                EventKind.LOCK_ACQUIRED.value,
                "DUT 与测试主机独占锁已获取",
            )
            await append_event(
                session,
                run,
                EventKind.RUN_STARTED.value,
                "模拟 Agent 已接管执行",
                target_state=RunState.RUNNING.value,
                projection={"started_at": utcnow()},
            )

        async with session_scope() as session:
            run = await _load_run(session, run_id)
            target = run.loop_target
            inject_failure = bool(run.params.get("inject_failure", False))
        failure_loop = min(3, target) if inject_failure and target >= 2 else -1

        for loop_index in range(1, target + 1):
            while True:
                async with session_scope() as session:
                    run = await _load_run(session, run_id)
                    if run.state != RunState.PAUSED.value:
                        break
                await _sleep(0.2)
            async with session_scope() as session:
                run = await _load_run(session, run_id)
                if run.state in {
                    RunState.ABORTED.value,
                    RunState.FROZEN.value,
                    RunState.REJECTED.value,
                }:
                    return
                await append_event(
                    session,
                    run,
                    EventKind.LOOP_STARTED.value,
                    "循环 {0}/{1} 开始".format(loop_index, target),
                    payload={"loop_index": loop_index},
                )
            await _sleep(0.22)

            if loop_index == failure_loop:
                async with session_scope() as session:
                    run = await _load_run(session, run_id)
                    await append_event(
                        session,
                        run,
                        EventKind.HEARTBEAT_LOST.value,
                        "Agent 心跳丢失；OOB 仍在线，启动自动恢复",
                        source=EventSource.OOB.value,
                        severity=Severity.ERROR.value,
                        payload={"loop_index": loop_index, "oob_online": True},
                        target_state=RunState.RECOVERING.value,
                    )
                await _sleep(0.25)
                await _run_recovery(run_id, loop_index)
                async with session_scope() as session:
                    run = await _load_run(session, run_id)
                    await append_event(
                        session,
                        run,
                        EventKind.CHECKPOINT_RESUMED.value,
                        "主机指纹一致，从循环检查点续跑",
                        payload={"loop_index": loop_index},
                    )

            async with session_scope() as session:
                run = await _load_run(session, run_id)
                checkpoint = {
                    "loop_index": loop_index,
                    "next_step_key": "regression-loop",
                    "dut_fw": "B",
                }
                await append_event(
                    session,
                    run,
                    EventKind.LOOP_COMPLETED.value,
                    "循环 {0}/{1} 通过，检查点已写入".format(loop_index, target),
                    payload={"loop_index": loop_index},
                    projection={"loop_done": loop_index, "checkpoint": checkpoint},
                )
            await _sleep(0.14)

        async with session_scope() as session:
            run = await _load_run(session, run_id)
            bundle = await build_evidence_bundle(session, run)
            await append_event(
                session,
                run,
                EventKind.EVIDENCE_BUNDLED.value,
                "Evidence Bundle 已生成，完整率 100%",
                payload={"bundle_id": bundle.id, "uri": bundle.uri, "completeness": 1.0},
            )
            await append_event(
                session,
                run,
                EventKind.LOCK_RELEASED.value,
                "DUT 与测试主机独占锁已释放",
            )
            await session.execute(delete(ResourceLock).where(ResourceLock.run_id == run.id))
            await append_event(
                session,
                run,
                EventKind.RUN_COMPLETED.value,
                "全部循环完成，结论 PASS",
                target_state=RunState.COMPLETED.value,
                projection={"verdict": Verdict.PASS.value, "ended_at": utcnow()},
            )
    except asyncio.CancelledError:
        raise
    finally:
        _tasks.pop(run_id, None)


def start_simulation(run_id: str) -> None:
    existing = _tasks.get(run_id)
    if existing and not existing.done():
        return
    task = asyncio.create_task(simulate_run(run_id))
    _tasks[run_id] = task
    _task_refs.add(task)
    task.add_done_callback(_task_refs.discard)


def cancel_simulation(run_id: str) -> None:
    task = _tasks.pop(run_id, None)
    if task and not task.done():
        task.cancel()
