"""Kernel Run 循环骨架。

事件序列对应 docs/architecture/lingtai-runtime.md §3.1；瀑布节点可被插件否决。
投影当前为「调用方组装」形态：prompt 由领域层按镜树祖先链组装后传入，
三类投影来源（祖先路径/获准记忆/显式引文）的完整接入随 M3 息壤落地。
``RunOutcome`` 携带终态与生成内容，供领域层（镜树节点固化等）在 Run 结束时回调。
"""

from __future__ import annotations

import asyncio
from collections.abc import Awaitable, Callable
from dataclasses import dataclass, field, replace
from datetime import UTC, datetime
from typing import Any

from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

from app.core.ids import new_trace_id
from app.kernel.eventlog import KernelEventLog
from app.kernel.invariants import (
    InvariantViolation,
    assert_model_visible_logged,
    assert_stream_reconstruction,
)
from app.kernel.seat import SeatManager
from app.models.runtime import Run
from app.runtime.bus import (
    NODE_LLM_PRE_REQUEST,
    NODE_RESULT_PRE_ACCEPT,
    NODE_RUN_REQUESTED,
    NODE_TURN_STARTED,
    RunContext,
    Veto,
    WaterfallBus,
)
from app.runtime.profile import Profile, resolve_profile
from app.shenshi_client.client import ShenshiClient
from app.shenshi_client.types import ShenshiChatRequest, ShenshiMessage

_SYSTEM_CONTRACT_PLACEHOLDER = "你是魂镜人物运行时（骨架占位契约）。"

# Run 结束回调：(生成内容, 终态)。终态仅 completed / interrupted。
OnRunFinal = Callable[[str, str], Awaitable[None]]


@dataclass
class RunOutcome:
    """Run 终态载体：循环内各终止路径写入，execute 返回前触发回调。"""

    status: str = "running"  # running / completed / interrupted / cancelled
    content: str = ""
    error_code: str = ""
    extras: dict[str, Any] = field(default_factory=dict)


class KernelRunLoop:
    """单次 Run 的执行骨架：命台席位 → 事件序列 → 神识流 → 接纳断言。

    ``character_id`` 允许为空（照心试演无正式人物），此时必须提供 ``seat_key``。
    """

    def __init__(
        self,
        *,
        event_log: KernelEventLog,
        seats: SeatManager,
        bus: WaterfallBus,
        shenshi: ShenshiClient,
        session_factory: async_sessionmaker[AsyncSession],
        timeout_seconds: float,
    ) -> None:
        self._log = event_log
        self._seats = seats
        self._bus = bus
        self._shenshi = shenshi
        self._sessions = session_factory
        self._timeout = timeout_seconds

    async def execute(
        self,
        *,
        run_id: str,
        character_id: str | None,
        prompt: str,
        trace_id: str | None = None,
        seat_key: str | None = None,
        system_contract: str | None = None,
        on_final: OnRunFinal | None = None,
        temperature: float | None = None,
        max_tokens: int | None = None,
    ) -> RunOutcome:
        assert character_id or seat_key, "Run 必须绑定人物或提供席位键（试演）"
        trace = trace_id or new_trace_id()
        outcome = RunOutcome()
        try:
            async with asyncio.timeout(self._timeout):
                await self._execute(
                    run_id=run_id,
                    character_id=character_id,
                    prompt=prompt,
                    trace_id=trace,
                    seat_key=seat_key,
                    system_contract=system_contract,
                    outcome=outcome,
                    temperature=temperature,
                    max_tokens=max_tokens,
                )
        except TimeoutError:
            outcome.status = "interrupted"
            outcome.error_code = "run_timeout"
            await self._finish_interrupted(
                run_id, character_id, trace, code="run_timeout", message="运行超时被中断"
            )
        except asyncio.CancelledError:
            outcome.status = "cancelled"
            await self._emit(run_id, character_id, trace, "run.cancelled", visibility="public")
            await self._update_run(run_id, status="cancelled")
            raise

        if on_final is not None and outcome.status in ("completed", "interrupted"):
            await on_final(outcome.content, outcome.status)
        return outcome

    async def _execute(
        self,
        *,
        run_id: str,
        character_id: str | None,
        prompt: str,
        trace_id: str,
        seat_key: str | None,
        system_contract: str | None,
        outcome: RunOutcome,
        temperature: float | None = None,
        max_tokens: int | None = None,
    ) -> None:
        ctx = RunContext(run_id=run_id, character_id=character_id, trace_id=trace_id)
        try:
            await self._bus.fire(NODE_RUN_REQUESTED, ctx)
        except Veto as veto:
            outcome.status, outcome.error_code = "interrupted", "HJ-1301"
            await self._record_veto(run_id, character_id, trace_id, veto)
            await self._finish_interrupted(
                run_id, character_id, trace_id, code="HJ-1301", message=veto.reason
            )
            return

        await self._emit(run_id, character_id, trace_id, "run.queued")
        async with self._seats.foreground(seat_key or character_id or run_id):
            await self._update_run(run_id, status="running")
            await self._emit(run_id, character_id, trace_id, "run.started")
            await self._turn(
                run_id,
                character_id,
                trace_id,
                prompt,
                system_contract or _SYSTEM_CONTRACT_PLACEHOLDER,
                outcome,
                temperature=temperature,
                max_tokens=max_tokens,
            )

    async def _turn(
        self,
        run_id: str,
        character_id: str | None,
        trace_id: str,
        prompt: str,
        system_contract: str,
        outcome: RunOutcome,
        temperature: float | None = None,
        max_tokens: int | None = None,
    ) -> None:
        ctx = RunContext(
            run_id=run_id, character_id=character_id, trace_id=trace_id, payload={"prompt": prompt}
        )
        profile: Profile = resolve_profile()
        await self._emit(run_id, character_id, trace_id, "turn.started")
        await self._bus.fire(NODE_TURN_STARTED, ctx)

        request = ShenshiChatRequest(
            model=profile.params.get("model"),
            slot=str(profile.params.get("slot", "chat")),
            messages=[
                ShenshiMessage(role="system", content=system_contract),
                ShenshiMessage(role="user", content=prompt),
            ],
            run_id=run_id,
            trace_id=trace_id,
            character_id=character_id,
            temperature=temperature,
            max_tokens=max_tokens,
        )
        await self._emit(
            run_id,
            character_id,
            trace_id,
            "context.projected",
            payload={
                "profile": profile.snapshot(),
                "sources": {"system_contract": "assembled", "user_input": True},
            },
            visibility="internal",
        )
        try:
            await self._bus.fire(NODE_LLM_PRE_REQUEST, ctx)
        except Veto as veto:
            outcome.status, outcome.error_code = "interrupted", "HJ-1301"
            await self._record_veto(run_id, character_id, trace_id, veto)
            await self._finish_interrupted(
                run_id, character_id, trace_id, code="HJ-1301", message=veto.reason
            )
            return

        # "模型可见即可重建"锚点：完整 IR 请求进事件日志
        await self._emit(
            run_id,
            character_id,
            trace_id,
            "llm.requested",
            payload={"request": request.model_dump(mode="json")},
            visibility="internal",
        )

        text_parts: list[str] = []
        usage: dict[str, Any] = {}
        try:
            async for delta in self._shenshi.stream_chat(request):
                if delta.delta:
                    text_parts.append(delta.delta)
                    await self._emit(
                        run_id, character_id, trace_id, "llm.delta", payload={"delta": delta.delta}
                    )
                if delta.finish_reason:
                    usage = {
                        "prompt_tokens": delta.usage_prompt_tokens,
                        "completion_tokens": delta.usage_completion_tokens,
                        "finish_reason": delta.finish_reason,
                    }
        except Exception as exc:  # 神识侧失败：如实上报，不伪装完成
            outcome.status, outcome.error_code = "interrupted", "HJ-2001"
            await self._emit(
                run_id,
                character_id,
                trace_id,
                "llm.failed",
                payload={"error": str(exc)},
                visibility="internal",
            )
            await self._finish_interrupted(
                run_id, character_id, trace_id, code="HJ-2001", message=f"神识模型调用失败：{exc}"
            )
            return

        await self._emit(run_id, character_id, trace_id, "llm.completed", payload={"usage": usage})

        accepted_text = "".join(text_parts)
        await self._bus.fire(
            NODE_RESULT_PRE_ACCEPT, replace(ctx, payload={"result": accepted_text})
        )
        events = await self._log.list_by_run(run_id)
        try:
            assert_model_visible_logged(request, events)
            assert_stream_reconstruction(events, accepted_text)
        except InvariantViolation as violation:
            outcome.status, outcome.error_code = "interrupted", "HJ-9000"
            await self._emit(
                run_id,
                character_id,
                trace_id,
                "run.invariant_violation",
                payload={"invariant": violation.invariant, "detail": violation.detail},
                visibility="audit",
            )
            await self._finish_interrupted(
                run_id,
                character_id,
                trace_id,
                code="HJ-9000",
                message=f"Kernel 不变量违反：{violation.invariant}",
            )
            return

        await self._emit(
            run_id, character_id, trace_id, "result.accepted", payload={"content": accepted_text}
        )
        await self._emit(run_id, character_id, trace_id, "turn.finished")
        await self._emit(run_id, character_id, trace_id, "run.finished")
        await self._update_run(run_id, status="completed")
        outcome.status = "completed"
        outcome.content = accepted_text

    async def _record_veto(
        self, run_id: str, character_id: str | None, trace_id: str, veto: Veto
    ) -> None:
        await self._emit(
            run_id,
            character_id,
            trace_id,
            "safety.ruled",
            payload={"reason": veto.reason, "rule": veto.rule, "details": veto.details},
            visibility="audit",
        )

    async def _finish_interrupted(
        self,
        run_id: str,
        character_id: str | None,
        trace_id: str,
        *,
        code: str,
        message: str,
    ) -> None:
        await self._emit(run_id, character_id, trace_id, "run.interrupted")
        await self._update_run(run_id, status="interrupted", error_code=code, error_message=message)

    async def _emit(
        self,
        run_id: str,
        character_id: str | None,
        trace_id: str,
        type: str,  # noqa: A002
        payload: dict[str, Any] | None = None,
        visibility: str = "public",
    ) -> None:
        await self._log.emit(
            type=type,
            run_id=run_id,
            character_id=character_id,
            trace_id=trace_id,
            payload=payload,
            visibility=visibility,
        )

    async def _update_run(
        self,
        run_id: str,
        *,
        status: str,
        error_code: str | None = None,
        error_message: str | None = None,
    ) -> None:
        async with self._sessions() as session:
            run = await session.get(Run, run_id)
            if run is None:
                return
            run.status = status
            run.error_code = error_code
            run.error_message = error_message
            if status == "running":
                run.started_at = datetime.now(UTC)
            if status in ("completed", "failed", "cancelled", "interrupted"):
                run.completed_at = datetime.now(UTC)
            await session.commit()