"""司命台管理聚合服务（api-spec §19.1–§19.3）。

九枢健康度由各枢运行/治理数据派生（不落长期状态，§19.1）；朱批待办复用
待决请命（§19.2）；起居注审计复用 audit_events（§19.3）。
"""

import json

from sqlalchemy import func, select

from app.core.config import settings
from app.core.security import redact_text
from app.core.time import to_iso, utc_now
from app.db.session import AsyncSessionLocal
from app.models.idempotency import IdempotencyKey
from app.models.xirang import Memory
from app.models.runtime import Event, Run
from app.models.safety import AuditEvent, PendingConfirmationRecord
from app.repositories import audit as audit_repo
from app.repositories import characters as character_repo
from app.repositories import idempotency as idempotency_repo
from app.repositories import runtime as runtime_repo
from app.schemas.admin import (
    AuditEntry,
    PendingApproval,
    PivotHealthReport,
    PivotHealthSummary,
    TraceDetail,
    TraceItem,
    TraceListResponse,
    TraceSummary,
)
from app.schemas.common import Page
from app.services.serializers import audit_to_response

# 九枢全集（契约字典 §5.9）：枢标识、中文名、工程模块名。
_PIVOT_REGISTRY = [
    ("personality_core", "灵核枢", "Personality Core"),
    ("capability_registry", "百工枢", "Capability Registry"),
    ("safety_governor", "司命枢", "Safety Governor"),
    ("hetu", "河图枢", "Hetu World"),
    ("tiangong", "天工阁", "Tiangong Backstage"),
    ("kouxin", "叩心枢", "Kouxin Proactive"),
    ("memory", "息壤枢", "Xirang Memory"),
]


def _summarize_input(value: dict | None) -> str:
    if not value:
        return ""
    text = json.dumps(value, ensure_ascii=False, default=str)
    return redact_text(text)[:200]


async def get_pivots_health(
    *,
    owner_scope: str = settings.default_owner_scope,
):
    """九枢健康度：以事件流与域表计数呈现（旧枢治理表随 M8 退役）。"""
    from app.domains.baigong.models import Capability
    from app.domains.hetu.models import PersistentWorld
    from app.domains.kouxin.models import KnockingCandidate
    from app.domains.tiangong.models import BackstageTask
    from app.models.safety import PendingConfirmationRecord
    from app.models.xirang import Memory

    async with AsyncSessionLocal() as session:
        pending_count = len(
            list(
                (
                    await session.execute(
                        select(PendingConfirmationRecord).where(
                            PendingConfirmationRecord.status == "pending"
                        )
                    )
                )
                .scalars()
                .all()
            )
        )
        caps_n = await _count(session, Capability)
        world_n = await _count(session, PersistentWorld)
        tasks_n = await _count(session, BackstageTask)
        knocks_n = await _count(session, KnockingCandidate)
        memories_n = await _count(session, Memory, Memory.status != "tombstoned")
        character = await character_repo.get_active_character(session, owner_scope)
        has_personality = bool(character and character.identity_baseline_json not in ("", "{}"))
        now = to_iso(utc_now())

        raw: list[tuple[str, str, str]] = [
            ("personality_core", "steady" if has_personality else "void", "道体锚点已就绪" if has_personality else "锚点未初始化"),
            ("capability_registry", "steady" if caps_n else "void", f"能力 {caps_n} 项"),
            ("safety_governor", "cracked" if pending_count else "steady", f"{pending_count} 项待裁决" if pending_count else "无待裁决"),
            ("hetu", "steady" if world_n else "void", f"境界 {world_n} 个"),
            ("tiangong", "steady" if tasks_n else "void", f"受托任务 {tasks_n} 项"),
            ("kouxin", "breathing" if knocks_n else "void", f"叩心候选 {knocks_n} 条"),
            ("memory", "steady" if memories_n else "void", f"记忆 {memories_n} 条"),
        ]

        name_map = {pivot: (name, module) for pivot, name, module in _PIVOT_REGISTRY}
        pivots = [
            PivotHealthSummary(
                pivot=pivot, name=name_map[pivot][0], module=name_map[pivot][1],
                health=health, summary=summary, updated_at=now,
            )
            for pivot, health, summary in raw
        ]
        alerts = sum(1 for item in pivots if item.health == "cracked")
        return PivotHealthReport(
            overall="cracked" if alerts else "steady", alerts=alerts, pivots=pivots
        )


async def _count(session, model, *criteria) -> int:
    from sqlalchemy import func as _func

    stmt = select(_func.count()).select_from(model)
    for c in criteria:
        stmt = stmt.where(c)
    return int((await session.execute(stmt)).scalar_one())

async def get_pending_approvals() -> list[PendingApproval]:
    async with AsyncSessionLocal() as session:
        records = (
            (
                await session.execute(
                    select(PendingConfirmationRecord)
                    .where(PendingConfirmationRecord.status == "pending")
                    .order_by(PendingConfirmationRecord.created_at.asc())
                )
            )
            .scalars()
            .all()
        )
    approvals: list[PendingApproval] = []
    for record in records:
        try:
            input_payload = json.loads(record.input_json) if record.input_json else {}
        except ValueError:
            input_payload = {}
        approvals.append(
            PendingApproval(
                confirmation_id=record.id,
                character_id=record.character_id,
                tool=record.tool,
                risk_level=record.risk_level,
                permission_policy="ask",
                action_summary=f"调用能力 {record.tool}",
                input_summary=_summarize_input(input_payload),
                possible_impact=None,
                reason=record.reason,
                created_at=record.created_at.isoformat() if record.created_at else "",
            )
        )
    return approvals


async def get_audit_timeline(
    *,
    pivot: str | None = None,
    page: int = 1,
    page_size: int = 50,
) -> Page[AuditEntry]:
    async with AsyncSessionLocal() as session:
        events, total = await audit_repo.list_audit_events(
            session, pivot=pivot, page=page, page_size=page_size
        )
        return Page(
            items=[audit_to_response(event) for event in events],
            page=page,
            page_size=page_size,
            total=total,
        )


# —— §19.5 链路追踪（按 trace_id 聚合 events / audit_events / idempotency_keys）——


def _event_to_item(event: Event) -> TraceItem:
    try:
        payload = json.loads(event.payload_json) if event.payload_json else {}
    except (json.JSONDecodeError, TypeError):
        payload = {}
    return TraceItem(
        kind="event",
        id=event.id,
        created_at=event.created_at,
        source_label=f"运行事件 / {event.type}",
        summary=event.type,
        detail={
            "run_id": event.run_id,
            "source": event.source,
            "type": event.type,
            "payload": payload,
            "sequence": event.sequence,
            "parent_id": event.parent_id,
        },
    )


def _audit_to_item(audit: AuditEvent) -> TraceItem:
    parts = [audit.decision, audit.risk_level]
    if audit.reason:
        parts.append(audit.reason[:60])
    return TraceItem(
        kind="audit",
        id=audit.id,
        created_at=audit.created_at,
        source_label=f"安全审计 / {audit.action}",
        summary=" · ".join(parts),
        detail={
            "pivot": audit.pivot,
            "action": audit.action,
            "actor": audit.actor,
            "decision": audit.decision,
            "risk_level": audit.risk_level,
            "reason": audit.reason,
            "resource_type": audit.resource_type,
            "resource_id": audit.resource_id,
        },
    )


def _idempotency_to_item(idem: IdempotencyKey) -> TraceItem:
    return TraceItem(
        kind="idempotency",
        id=idem.id,
        created_at=idem.created_at,
        source_label=f"请求幂等 / {idem.method}",
        summary=f"{idem.method} {idem.path} → {idem.response_status_code}",
        detail={
            "method": idem.method,
            "path": idem.path,
            "status": idem.status,
            "response_status_code": idem.response_status_code,
            "scope": idem.scope,
        },
    )


async def get_trace(trace_id: str) -> TraceDetail:
    """按 trace_id 聚合 events / audit_events / idempotency_keys 三表为升序时间线。"""
    async with AsyncSessionLocal() as session:
        events = await runtime_repo.list_events_by_trace(session, trace_id)
        audits = await audit_repo.list_audit_events_by_trace(session, trace_id)
        idempotencies = await idempotency_repo.list_by_trace(session, trace_id)

        items: list[TraceItem] = [_event_to_item(event) for event in events]
        items.extend(_audit_to_item(audit) for audit in audits)
        items.extend(_idempotency_to_item(idem) for idem in idempotencies)
        items.sort(key=lambda item: item.created_at)

        run_ids = sorted({event.run_id for event in events if event.run_id})
        character_ids = sorted(
            {
                *[event.character_id for event in events if event.character_id],
                *[audit.character_id for audit in audits if audit.character_id],
            }
        )
        return TraceDetail(
            trace_id=trace_id,
            character_ids=character_ids,
            run_ids=run_ids,
            session_ids=[],
            items=items,
        )


async def list_time_events(*, limit: int = 100, offset: int = 0) -> tuple[list[Event], int]:
    """列出辰正 time.* 事件（source='time_sense'），供起居注时间河展示（设计 §8 司命台）。"""
    async with AsyncSessionLocal() as session:
        result = await session.execute(
            select(Event)
            .where(Event.source == "time_sense")
            .order_by(Event.created_at.desc())
            .limit(limit)
            .offset(offset)
        )
        items = list(result.scalars().all())
        count_result = await session.execute(
            select(func.count()).select_from(Event).where(Event.source == "time_sense")
        )
        total = int(count_result.scalar_one())
    return items, total


async def list_traces(*, limit: int = 50) -> TraceListResponse:
    """列出最近活动的 trace_id 概览（跨 events / audit_events / idempotency_keys 聚合），
    含每条链路涉及的人物与对话归属。"""
    async with AsyncSessionLocal() as session:
        merged: dict[str, dict[str, object]] = {}

        def _entry(tid: str) -> dict[str, object]:
            return merged.setdefault(
                tid,
                {
                    "trace_id": tid,
                    "last_seen": None,
                    "event_count": 0,
                    "audit_count": 0,
                    "idempotency_count": 0,
                    "character_ids": set(),
                    "session_ids": set(),
                },
            )

        def _absorb(stats: list[tuple[str, int, object]], count_key: str) -> None:
            for tid, cnt, last in stats:
                entry = _entry(tid)
                entry[count_key] = cnt
                if last is not None and (entry["last_seen"] is None or last > entry["last_seen"]):
                    entry["last_seen"] = last

        _absorb(await runtime_repo.summarize_by_trace(session), "event_count")
        _absorb(await audit_repo.summarize_by_trace(session), "audit_count")
        _absorb(await idempotency_repo.summarize_by_trace(session), "idempotency_count")

        # 人物归属（events + audits 的 character_id）
        for tid, cid in (
            await session.execute(
                select(Event.trace_id, Event.character_id).where(
                    Event.trace_id.is_not(None), Event.character_id.is_not(None)
                )
            )
        ).all():
            _entry(tid)["character_ids"].add(cid)
        for tid, cid in (
            await session.execute(
                select(AuditEvent.trace_id, AuditEvent.character_id).where(
                    AuditEvent.trace_id.is_not(None),
                    AuditEvent.character_id.is_not(None),
                )
            )
        ).all():
            _entry(tid)["character_ids"].add(cid)

        # 对话归属随线性会话模型退役；镜树归属随 M2c 司命台重设计接入（tree_id 聚合）

        total = len(merged)
        ordered = sorted(merged.values(), key=lambda e: e["last_seen"], reverse=True)
        items = [
            TraceSummary(
                trace_id=str(entry["trace_id"]),
                last_seen=entry["last_seen"],  # type: ignore[arg-type]
                character_ids=sorted(entry["character_ids"]),  # type: ignore[arg-type]
                session_ids=sorted(entry["session_ids"]),  # type: ignore[arg-type]
                event_count=int(entry["event_count"]),
                audit_count=int(entry["audit_count"]),
                idempotency_count=int(entry["idempotency_count"]),
            )
            for entry in ordered[:limit]
        ]
        return TraceListResponse(items=items, total=total)