wecom_it_smart_desk/backend/app/services/automation/session_manager.py

# =============================================================================
# 企微IT智能服务台 — 阶段5 自动化 会话编排服务
# =============================================================================
# 说明：自动化会话的编排中枢，串联「意图识别 → 场景校验 → 终端映射 →
#       动作计划生成 → 执行引擎」。同时提供会话 CRUD、转人工、结果反馈、
#       静默关单等接口。
#
# 编排在后台任务中运行（run_session_in_background），API 创建会话后立即返回，
# 进度通过 WS 实时推送。
# =============================================================================

from __future__ import annotations

import asyncio
import logging
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional

from sqlalchemy import select

from app.config import settings
from app.constants import (
    AutomationErrorCode,
    GLOBAL_INTENT_CORRECT,
    GLOBAL_INTENT_PAUSE,
    GLOBAL_INTENT_RESUME_TASK,
    GLOBAL_INTENT_SUPPLEMENT,
    PAUSE_TIMEOUT_HOURS,
    REDIS_KEY_PAUSED_SESSIONS,
    REDIS_KEY_RESUME_POINT,
    RESUME_POINT_REDIS_TTL,
)
from app.database import _get_session_factory
from app.models.automation import (
    ApprovalTicket,
    AutoAction,
    AutoSession,
    InformationItem,
    ScenarioConfig,
)
from app.services.automation.approval import ApprovalService
from app.services.automation.exception_handler import AutomationException
from app.services.automation.executor import ActionExecutor
from app.services.automation.information_item_service import InformationItemService
from app.services.automation.intent_router import IntentRouter
from app.services.automation.mapping_resolver import MappingResolver
from app.services.automation.progress_publisher import (
    cancel_silent_close,
    publish_info_corrected,
    publish_info_supplemented,
    publish_paused,
    publish_progress,
    publish_resumed,
    publish_takeover,
    publish_timeout_closed,
)
from app.services.automation import DEFAULT_SCENARIO_CONFIGS

logger = logging.getLogger(__name__)

# 需要终端映射的场景
_SCENARIOS_NEED_TERMINAL = {"virus_dispose", "terminal_locate"}


class AutoSessionService:
    """自动化会话编排服务。"""

    def __init__(self, db: Any, redis: Any = None, audit: Any = None):
        self.db = db
        self.redis = redis
        self.audit = audit

    # --------------------------------------------------------------------------
    # 会话 CRUD
    # --------------------------------------------------------------------------
    async def create_session(
        self,
        conversation_id: Optional[str],
        employee_id: str,
        description: str,
        mode: str = "real_exec",
    ) -> AutoSession:
        """创建自动化会话（状态=created）。"""
        session = AutoSession(
            conversation_id=conversation_id,
            employee_id=employee_id,
            mode=mode,
            title=(description or "")[:200],
            status="created",
            meta={"description": description},
        )
        self.db.add(session)
        await self.db.flush()
        return session

    async def get_session(self, session_id: str) -> Optional[AutoSession]:
        """按 ID 取会话。"""
        stmt = select(AutoSession).where(AutoSession.id == session_id)
        return (await self.db.execute(stmt)).scalar_one_or_none()

    async def list_sessions(
        self,
        employee_id: Optional[str] = None,
        agent_id: Optional[str] = None,
        status: Optional[str] = None,
        page: int = 1,
        page_size: int = 50,
    ) -> List[AutoSession]:
        """列出会话（支持过滤分页）。"""
        stmt = select(AutoSession)
        if employee_id:
            stmt = stmt.where(AutoSession.employee_id == employee_id)
        if agent_id:
            stmt = stmt.where(AutoSession.agent_id == agent_id)
        if status:
            stmt = stmt.where(AutoSession.status == status)
        stmt = stmt.order_by(AutoSession.created_at.desc())
        stmt = stmt.limit(page_size).offset((page - 1) * page_size)
        return list((await self.db.execute(stmt)).scalars().all())

    async def get_session_detail(
        self, session_id: str
    ) -> Optional[Dict[str, Any]]:
        """取会话详情（含动作列表与当前待决审批单）。"""
        session = await self.get_session(session_id)
        if session is None:
            return None
        act_stmt = select(AutoAction).where(AutoAction.session_id == session_id)
        actions = list((await self.db.execute(act_stmt)).scalars().all())
        actions.sort(key=lambda a: a.action_index)

        ticket = None
        if session.current_action_id:
            tstmt = select(ApprovalTicket).where(
                ApprovalTicket.action_id == session.current_action_id,
                ApprovalTicket.status == "pending",
            )
            ticket = (await self.db.execute(tstmt)).scalar_one_or_none()
        return {"session": session, "actions": actions, "ticket": ticket}

    # --------------------------------------------------------------------------
    # 编排主流程
    # --------------------------------------------------------------------------
    async def start(self, session_id: str) -> None:
        """编排：意图识别 → 全局意图分流 → 场景校验 → 映射 → 计划 → 执行。

        复杂场景重构：意图识别后先检查 global_intent，
        命中 pause/resume_task/correct/supplement 时分流到对应方法。
        """
        session = await self.get_session(session_id)
        if session is None:
            logger.warning(f"start 会话不存在: {session_id}")
            return
        if session.status != "created":
            logger.info(f"会话已启动过，跳过: {session_id} status={session.status}")
            return

        session.status = "running"
        await self.db.flush()
        await publish_progress(session.id, "start", "开始自动化处置")

        description = (session.meta or {}).get("description", "")

        # 1. 意图识别（含全局意图检测）
        router = IntentRouter(self.db, audit=self.audit)
        try:
            intent = await router.detect(description, session.employee_id)
        except Exception as e:  # noqa: BLE001
            intent = {"global_intent": None, "scenario_key": None, "confidence": 0.0, "error": str(e)}
        session.scenario_key = intent.get("scenario_key")
        session.confidence = float(intent.get("confidence") or 0.0)
        session.intent = intent
        await self.db.flush()
        await publish_progress(
            session.id,
            "intent",
            f"识别场景: {session.scenario_key or '未知'}（置信度 {session.confidence:.2f}）",
        )

        # 1.5 全局意图分流（复杂场景重构）
        global_intent = intent.get("global_intent")
        if global_intent == GLOBAL_INTENT_PAUSE:
            await self.pause_session(session_id, reason="用户主动暂停")
            return
        if global_intent == GLOBAL_INTENT_RESUME_TASK:
            # 当前会话刚创建，恢复逻辑应指向已有暂停会话
            await self.resume_session(session.employee_id)
            return
        if global_intent == GLOBAL_INTENT_CORRECT:
            field = intent.get("corrected_field") or ""
            new_value = intent.get("new_value") or ""
            old_value = intent.get("old_value")
            if field and new_value:
                await self.correct_info(session_id, field, new_value, old_value)
            return
        if global_intent == GLOBAL_INTENT_SUPPLEMENT:
            field = intent.get("supplement_field") or ""
            value = intent.get("supplement_value") or ""
            if field and value:
                await self.supplement_info(session_id, field, value)
            return

        # 2. 置信度门槛 → 低置信度转人工
        thresholds = settings.get_automation_thresholds()
        confidence_min = float(thresholds.get("confidence_min", 0.6))
        if not session.scenario_key or session.confidence < confidence_min:
            await self._handoff(
                session, f"意图识别置信度不足（{session.confidence:.2f}），转人工"
            )
            return

        # 3. 场景配置校验
        scenario = await self._get_scenario_config(session.scenario_key)
        if scenario is None or not scenario.enabled:
            await self._handoff(
                session, f"场景未启用或未配置: {session.scenario_key}"
            )
            return

        # 4. 终端映射（按需）
        mapping: Dict[str, Any] = {}
        if session.scenario_key in _SCENARIOS_NEED_TERMINAL:
            resolver = MappingResolver(self.db, audit=self.audit)
            mapping = await resolver.resolve(session.employee_id, session.scenario_key)
            if (
                session.scenario_key == "virus_dispose"
                and not mapping.get("client_ids")
            ):
                await self._handoff(session, "未解析到目标终端，转人工")
                return
        session.meta = {**(session.meta or {}), "mapping": mapping}
        await self.db.flush()

        # 5. 生成动作计划
        plan = self._build_actions(scenario, mapping)
        for idx, item in enumerate(plan):
            action = AutoAction(session_id=session.id, action_index=idx, **item)
            self.db.add(action)
        await self.db.flush()
        await publish_progress(
            session.id, "plan_ready", f"已生成处置方案（共 {len(plan)} 步）"
        )

        # 6. 执行引擎
        executor = ActionExecutor(self.db, self.redis, audit=self.audit)
        await executor.run(session_id)

    def _build_actions(
        self, scenario: ScenarioConfig, mapping: Dict[str, Any]
    ) -> List[Dict[str, Any]]:
        """从场景配置（或默认模板）生成动作计划。"""
        plan_items = scenario.actions
        if not plan_items:
            default = DEFAULT_SCENARIO_CONFIGS.get(scenario.scenario_key, {})
            plan_items = default.get("actions", [])

        result: List[Dict[str, Any]] = []
        for it in plan_items:
            params: Dict[str, Any] = dict(it.get("params") or {})
            confirm_channel = it.get("confirm_channel")
            if mapping.get("client_ids"):
                params.setdefault("client_ids", mapping["client_ids"])
            if confirm_channel:
                params["confirm_channel"] = confirm_channel
            result.append(
                {
                    "action_type": it.get("action_type", ""),
                    "adapter": it.get("adapter", "internal"),
                    "risk_level": it.get("risk_level", "read"),
                    "title": it.get("title", it.get("action_type", "")),
                    "description": it.get("description", it.get("title", "")),
                    "payload": params,
                }
            )
        return result

    async def _get_scenario_config(
        self, scenario_key: str
    ) -> Optional[ScenarioConfig]:
        """加载场景配置（DB 优先，缺失则用默认模板的开关）。"""
        stmt = select(ScenarioConfig).where(
            ScenarioConfig.scenario_key == scenario_key
        )
        config = (await self.db.execute(stmt)).scalar_one_or_none()
        if config is not None:
            return config
        # DB 无记录 → 用默认模板（默认启用）
        default = DEFAULT_SCENARIO_CONFIGS.get(scenario_key)
        if default is None:
            return None
        return ScenarioConfig(
            scenario_key=scenario_key,
            name=default.get("name", scenario_key),
            description=default.get("description", ""),
            enabled=default.get("enabled", True),
            trigger_conditions=default.get("trigger_conditions"),
            actions=default.get("actions"),
            approval_strategy=default.get("approval_strategy"),
        )

    async def _handoff(self, session: AutoSession, reason: str) -> None:
        """置为转人工并推送事件。"""
        session.status = "handoff"
        session.closed_by = "system(auto)"
        await self.db.flush()
        await publish_takeover(session.id, reason)

    # --------------------------------------------------------------------------
    # 转人工 / 反馈 / 关单
    # --------------------------------------------------------------------------
    async def takeover(
        self, session_id: str, agent_id: str, note: Optional[str] = None
    ) -> AutoSession:
        """转人工接管。"""
        session = await self.get_session(session_id)
        if session is None:
            raise AutomationException(AutomationErrorCode.SESSION_NOT_FOUND)
        session.status = "handoff"
        session.agent_id = agent_id
        session.closed_by = agent_id
        await self.db.flush()
        cancel_silent_close(session_id)
        await publish_takeover(session.id, f"坐席 {agent_id} 接管：{note or ''}")
        return session

    async def resolve_feedback(
        self, session_id: str, satisfied: bool, note: Optional[str] = None
    ) -> AutoSession:
        """员工处置结果反馈。

        满意 → 关单；不满意 → 转人工。
        """
        session = await self.get_session(session_id)
        if session is None:
            raise AutomationException(AutomationErrorCode.SESSION_NOT_FOUND)
        cancel_silent_close(session_id)
        if satisfied:
            session.status = "closed"
            session.closed_by = session.employee_id
        else:
            session.status = "handoff"
            session.closed_by = "employee(reject)"
            await publish_takeover(session.id, f"员工不满意，转人工：{note or ''}")
        await self.db.flush()
        return session

    async def auto_close(self, session_id: str) -> None:
        """静默关单（仅 resolved 态可关）。"""
        session = await self.get_session(session_id)
        if session is None:
            return
        if session.status != "resolved":
            return
        session.status = "closed"
        session.closed_by = "system(auto)"
        await self.db.flush()

    # --------------------------------------------------------------------------
    # 复杂场景重构：暂停 / 恢复 / 更正 / 补充 / 坐席操作
    # --------------------------------------------------------------------------
    async def pause_session(
        self, session_id: str, reason: Optional[str] = None
    ) -> AutoSession:
        """暂停会话。

        校验状态：running / await_approval → paused；终态不可暂停。
        构建恢复点快照存入 Redis，推送 WS 暂停事件。

        Args:
            session_id: 会话ID
            reason: 暂停原因
        Returns:
            AutoSession: 暂停后的会话
        Raises:
            AutomationException: 会话不存在或不可暂停
        """
        session = await self.get_session(session_id)
        if session is None:
            raise AutomationException(AutomationErrorCode.SESSION_NOT_FOUND)

        # 终态不可暂停
        if session.status in ("closed", "handoff", "error"):
            raise AutomationException(AutomationErrorCode.SESSION_NOT_PAUSABLE)

        # 如果是 await_approval 状态暂停 → 记录到 meta（挂起审批计时）
        meta = dict(session.meta or {})
        if session.status == "await_approval" or self._is_awaiting_approval(session):
            meta["paused_from_approval"] = True

        now = datetime.now(timezone.utc)
        session.status = "paused"
        session.paused_at = now
        session.meta = meta
        await self.db.flush()

        # 构建恢复点快照
        info_svc = InformationItemService(self.db, self.redis)
        info_items = await info_svc.get_items(session_id)
        info_items_snapshot = [
            {
                "id": item.id,
                "name": item.name,
                "value": item.value,
                "modifiers": item.modifiers or [],
                "is_filled": item.is_filled,
                "is_locked": item.is_locked,
                "version": item.version,
            }
            for item in info_items
        ]
        step_desc = await self._get_current_step_desc(session)
        resume_point = {
            "title": session.title,
            "scenario_key": session.scenario_key,
            "current_action_id": session.current_action_id,
            "step_desc": step_desc,
            "info_items": info_items_snapshot,
            "paused_at": now.isoformat(),
        }

        # Redis 存储恢复点 + 暂停会话集合
        if self.redis:
            try:
                import json
                resume_key = REDIS_KEY_RESUME_POINT.format(session_id=session_id)
                await self.redis.setex(
                    resume_key,
                    RESUME_POINT_REDIS_TTL,
                    json.dumps(resume_point, ensure_ascii=False),
                )
                paused_key = REDIS_KEY_PAUSED_SESSIONS.format(
                    employee_id=session.employee_id
                )
                await self.redis.sadd(paused_key, session_id)
            except Exception as e:  # noqa: BLE001
                logger.warning(f"Redis 存储恢复点失败 session={session_id}: {e}")

        # 推送 WS 暂停事件
        resume_hint = "需要继续时跟我说一声「继续」就好"
        await publish_paused(
            session_id=session_id,
            title=session.title,
            paused_at=now.isoformat(),
            resume_hint=resume_hint,
        )

        logger.info(f"暂停会话: session={session_id} reason={reason or ''}")
        return session

    async def resume_session(
        self,
        employee_id: str,
        session_id: Optional[str] = None,
    ) -> Dict[str, Any]:
        """恢复暂停的会话。

        若 session_id 为 None → 查询该员工的暂停会话列表。
        若多个暂停会话 → 返回列表供前端选择（不直接恢复）。
        若单个 → 直接恢复。

        Args:
            employee_id: 员工ID
            session_id: 指定恢复的会话ID（可选）
        Returns:
            Dict: {"session": AutoSession, "resume_point": dict, "need_select": bool, "paused_list": list}
        Raises:
            AutomationException: 会话不存在或不可恢复
        """
        # 未指定 session_id → 查询暂停会话列表
        if session_id is None:
            paused_list = await self.list_paused_sessions(employee_id)
            if len(paused_list) == 0:
                return {"session": None, "resume_point": None, "need_select": False, "paused_list": []}
            if len(paused_list) > 1:
                return {
                    "session": None,
                    "resume_point": None,
                    "need_select": True,
                    "paused_list": paused_list,
                }
            # 单个暂停会话 → 直接恢复
            session_id = paused_list[0]["session_id"]

        session = await self.get_session(session_id)
        if session is None:
            raise AutomationException(AutomationErrorCode.SESSION_NOT_FOUND)

        if session.status != "paused":
            if session.status == "closed":
                raise AutomationException(
                    AutomationErrorCode.SESSION_NOT_RESUMABLE,
                    "该任务已超时关闭，请重新发起",
                )
            raise AutomationException(AutomationErrorCode.SESSION_NOT_RESUMABLE)

        # 从 Redis 加载恢复点
        resume_point = await self._load_resume_point(session_id)

        # 恢复状态
        now = datetime.now(timezone.utc)
        session.status = "running"
        session.paused_at = None
        await self.db.flush()

        # Redis 清理
        if self.redis:
            try:
                resume_key = REDIS_KEY_RESUME_POINT.format(session_id=session_id)
                await self.redis.delete(resume_key)
                paused_key = REDIS_KEY_PAUSED_SESSIONS.format(
                    employee_id=employee_id
                )
                await self.redis.srem(paused_key, session_id)
            except Exception as e:  # noqa: BLE001
                logger.warning(f"Redis 清理恢复点失败 session={session_id}: {e}")

        # 检查必需信息项完整性
        info_svc = InformationItemService(self.db, self.redis)
        pending_items = await info_svc.get_pending_required_items(session_id)

        # 推送 WS 恢复事件
        current_step = resume_point.get("step_desc", "") if resume_point else ""
        await publish_resumed(
            session_id=session_id,
            title=session.title,
            resumed_at=now.isoformat(),
            current_step=current_step,
        )

        # 若信息完整 → 续行执行
        if not pending_items:
            executor = ActionExecutor(self.db, self.redis, audit=self.audit)
            asyncio.create_task(self._run_executor(session_id))

        logger.info(f"恢复会话: session={session_id} pending_items={pending_items}")
        return {
            "session": session,
            "resume_point": resume_point,
            "need_select": False,
            "paused_list": [],
            "pending_items": pending_items,
        }

    async def list_paused_sessions(self, employee_id: str) -> List[Dict[str, Any]]:
        """获取员工的暂停会话列表。

        Args:
            employee_id: 员工ID
        Returns:
            List[Dict]: 暂停会话列表项（含暂停时长）
        """
        stmt = select(AutoSession).where(
            AutoSession.employee_id == employee_id,
            AutoSession.status == "paused",
        ).order_by(AutoSession.paused_at.desc())
        sessions = list((await self.db.execute(stmt)).scalars().all())

        now = datetime.now(timezone.utc)
        result: List[Dict[str, Any]] = []
        for s in sessions:
            paused_at = s.paused_at or s.updated_at
            duration = self._format_duration(paused_at, now) if paused_at else ""
            result.append({
                "session_id": s.id,
                "title": s.title,
                "scenario_key": s.scenario_key,
                "paused_at": paused_at.isoformat() if paused_at else None,
                "paused_duration": duration,
            })
        return result

    async def correct_info(
        self,
        session_id: str,
        field: str,
        new_value: str,
        old_value: Optional[str] = None,
    ) -> InformationItem:
        """信息更正。

        委托 InformationItemService.correct_value()，检查下游影响，
        推送 WS 更正事件。

        Args:
            session_id: 会话ID
            field: 更正的字段名
            new_value: 新值
            old_value: 旧值（可选）
        Returns:
            InformationItem: 更正后的信息项
        """
        info_svc = InformationItemService(self.db, self.redis)
        item = await info_svc.correct_value(session_id, field, new_value, old_value)

        # 检查下游影响
        has_impact = await info_svc.check_downstream_impact(session_id, field)
        if has_impact:
            logger.info(f"更正影响下游动作: session={session_id} field={field}")
            # TODO: 重新校验映射/动作计划（后续阶段实现）

        # 推送 WS 更正事件
        actual_old_value = old_value
        if not actual_old_value and item.update_history:
            actual_old_value = item.update_history[-1].get("old_value", "")
        await publish_info_corrected(
            session_id=session_id,
            field=field,
            old_value=actual_old_value or "",
            new_value=new_value,
            version=item.version,
        )

        logger.info(f"更正信息: session={session_id} field={field} v={item.version}")
        return item

    async def supplement_info(
        self,
        session_id: str,
        field: str,
        value: str,
    ) -> InformationItem:
        """信息补充。

        委托 InformationItemService.supplement_value()，推送 WS 补充事件。

        Args:
            session_id: 会话ID
            field: 补充的字段名
            value: 补充值
        Returns:
            InformationItem: 补充后的信息项
        """
        info_svc = InformationItemService(self.db, self.redis)
        item = await info_svc.supplement_value(session_id, field, value)

        # 推送 WS 补充事件
        await publish_info_supplemented(
            session_id=session_id,
            field=field,
            supplement_value=value,
            new_value=item.value,
        )

        logger.info(f"补充信息: session={session_id} field={field} v={item.version}")
        return item

    async def agent_resume(
        self,
        session_id: str,
        agent_id: str,
        note: Optional[str] = None,
    ) -> AutoSession:
        """坐席代恢复暂停会话。

        无需员工授权，恢复后标记 closed_by = "agent:{agent_id}(resume)"。

        Args:
            session_id: 会话ID
            agent_id: 坐席ID
            note: 备注
        Returns:
            AutoSession: 恢复后的会话
        Raises:
            AutomationException: 会话不存在或不可恢复
        """
        session = await self.get_session(session_id)
        if session is None:
            raise AutomationException(AutomationErrorCode.SESSION_NOT_FOUND)

        if session.status != "paused":
            raise AutomationException(AutomationErrorCode.SESSION_NOT_RESUMABLE)

        # 从 Redis 加载恢复点
        resume_point = await self._load_resume_point(session_id)

        now = datetime.now(timezone.utc)
        session.status = "running"
        session.paused_at = None
        session.agent_id = agent_id
        session.closed_by = f"agent:{agent_id}(resume)"
        await self.db.flush()

        # Redis 清理
        if self.redis:
            try:
                resume_key = REDIS_KEY_RESUME_POINT.format(session_id=session_id)
                await self.redis.delete(resume_key)
                paused_key = REDIS_KEY_PAUSED_SESSIONS.format(
                    employee_id=session.employee_id
                )
                await self.redis.srem(paused_key, session_id)
            except Exception as e:  # noqa: BLE001
                logger.warning(f"Redis 清理恢复点失败 session={session_id}: {e}")

        # 推送 WS 恢复事件
        current_step = resume_point.get("step_desc", "") if resume_point else ""
        await publish_resumed(
            session_id=session_id,
            title=session.title,
            resumed_at=now.isoformat(),
            current_step=f"[坐席代恢复] {current_step}",
        )

        # 续行执行
        asyncio.create_task(self._run_executor(session_id))

        logger.info(f"坐席代恢复: session={session_id} agent={agent_id} note={note or ''}")
        return session

    async def agent_close(
        self,
        session_id: str,
        agent_id: str,
        note: Optional[str] = None,
    ) -> AutoSession:
        """坐席手动关闭暂停会话。

        Args:
            session_id: 会话ID
            agent_id: 坐席ID
            note: 备注
        Returns:
            AutoSession: 关闭后的会话
        Raises:
            AutomationException: 会话不存在或不可关闭
        """
        session = await self.get_session(session_id)
        if session is None:
            raise AutomationException(AutomationErrorCode.SESSION_NOT_FOUND)

        if session.status != "paused":
            raise AutomationException(
                AutomationErrorCode.SESSION_NOT_RESUMABLE,
                "仅暂停状态的会话可由坐席关闭",
            )

        now = datetime.now(timezone.utc)
        session.status = "closed"
        session.closed_by = f"agent:{agent_id}(close)"
        session.agent_id = agent_id
        await self.db.flush()

        # Redis 清理
        if self.redis:
            try:
                resume_key = REDIS_KEY_RESUME_POINT.format(session_id=session_id)
                await self.redis.delete(resume_key)
                paused_key = REDIS_KEY_PAUSED_SESSIONS.format(
                    employee_id=session.employee_id
                )
                await self.redis.srem(paused_key, session_id)
            except Exception as e:  # noqa: BLE001
                logger.warning(f"Redis 清理恢复点失败 session={session_id}: {e}")

        logger.info(f"坐席关闭会话: session={session_id} agent={agent_id} note={note or ''}")
        return session

    # --------------------------------------------------------------------------
    # 复杂场景重构：内部辅助方法
    # --------------------------------------------------------------------------
    async def _load_resume_point(self, session_id: str) -> Optional[Dict[str, Any]]:
        """从 Redis 加载恢复点快照。"""
        if not self.redis:
            return None
        try:
            import json
            resume_key = REDIS_KEY_RESUME_POINT.format(session_id=session_id)
            raw = await self.redis.get(resume_key)
            if raw:
                return json.loads(raw)
        except Exception as e:  # noqa: BLE001
            logger.warning(f"加载恢复点失败 session={session_id}: {e}")
        return None

    async def _get_current_step_desc(self, session: AutoSession) -> str:
        """获取当前步骤描述（用于恢复点）。"""
        if not session.current_action_id:
            return "等待开始处置"
        stmt = select(AutoAction).where(AutoAction.id == session.current_action_id)
        action = (await self.db.execute(stmt)).scalar_one_or_none()
        if action:
            return f"当前步骤：{action.title}（{action.status}）"
        return "处置进行中"

    def _is_awaiting_approval(self, session: AutoSession) -> bool:
        """检查会话是否有待审批动作。"""
        # 通过 current_action_id 和 status 间接判断
        return session.current_action_id is not None and session.status == "paused"

    @staticmethod
    def _format_duration(start: datetime, end: datetime) -> str:
        """格式化时长为人类可读字符串（如 "2h 15min"）。"""
        # SQLite 读取的 datetime 可能是 timezone-naive，统一补上 UTC 时区后再相减
        if start.tzinfo is None:
            start = start.replace(tzinfo=timezone.utc)
        if end.tzinfo is None:
            end = end.replace(tzinfo=timezone.utc)
        delta = end - start
        total_seconds = int(delta.total_seconds())
        if total_seconds < 0:
            return ""
        hours = total_seconds // 3600
        minutes = (total_seconds % 3600) // 60
        if hours > 0:
            return f"{hours}h {minutes}min"
        return f"{minutes}min"

    async def _run_executor(self, session_id: str) -> None:
        """在独立 DB 会话中运行执行器（续行）。"""
        factory = _get_session_factory()
        async with factory() as db:
            svc = AutoSessionService(db, self.redis)
            executor = ActionExecutor(db, self.redis, audit=self.audit)
            try:
                await executor.run(session_id)
                await db.commit()
            except Exception as e:  # noqa: BLE001
                await db.rollback()
                logger.error(f"续行执行失败 session={session_id}: {e}")

    # --------------------------------------------------------------------------
    # 场景配置管理（管理端）
    # --------------------------------------------------------------------------
    async def list_scenario_configs(self) -> List[ScenarioConfig]:
        """列出全部场景配置。"""
        stmt = select(ScenarioConfig).order_by(ScenarioConfig.scenario_key)
        return list((await self.db.execute(stmt)).scalars().all())

    async def upsert_scenario_config(
        self, scenario_key: str, data: Dict[str, Any], operator: str = ""
    ) -> ScenarioConfig:
        """更新或创建场景配置，并快照为规则版本。"""
        from app.models.automation import RuleVersion

        stmt = select(ScenarioConfig).where(
            ScenarioConfig.scenario_key == scenario_key
        )
        config = (await self.db.execute(stmt)).scalar_one_or_none()
        if config is None:
            config = ScenarioConfig(scenario_key=scenario_key)
            self.db.add(config)

        if data.get("name") is not None:
            config.name = data["name"]
        if data.get("description") is not None:
            config.description = data["description"]
        if data.get("enabled") is not None:
            config.enabled = data["enabled"]
        if data.get("trigger_conditions") is not None:
            config.trigger_conditions = data["trigger_conditions"]
        if data.get("actions") is not None:
            config.actions = data["actions"]
        if data.get("approval_strategy") is not None:
            config.approval_strategy = data["approval_strategy"]

        await self.db.flush()

        # 快照为规则版本
        last = (
            await self.db.execute(
                select(RuleVersion.version)
                .where(RuleVersion.scenario_key == scenario_key)
                .order_by(RuleVersion.version.desc())
                .limit(1)
            )
        ).scalar_one_or_none()
        new_version = (last or 0) + 1
        snapshot = RuleVersion(
            scenario_key=scenario_key,
            version=new_version,
            content={
                "actions": config.actions,
                "approval_strategy": config.approval_strategy,
                "trigger_conditions": config.trigger_conditions,
            },
            status="published",
            canary_percent=100,
            created_by=operator or "admin",
            remark=f"更新场景配置至 v{new_version}",
        )
        self.db.add(snapshot)
        await self.db.flush()
        config.current_version_id = snapshot.id
        await self.db.flush()
        return config

    async def list_rule_versions(
        self, scenario_key: Optional[str] = None
    ) -> List[RuleVersion]:
        """列出规则版本。"""
        stmt = select(RuleVersion)
        if scenario_key:
            stmt = stmt.where(RuleVersion.scenario_key == scenario_key)
        stmt = stmt.order_by(RuleVersion.created_at.desc())
        return list((await self.db.execute(stmt)).scalars().all())

    async def metrics(self) -> Dict[str, Any]:
        """汇总看板指标。"""
        from sqlalchemy import func

        total = (
            await self.db.execute(select(func.count(AutoSession.id)))
        ).scalar() or 0
        resolved = (
            await self.db.execute(
                select(func.count(AutoSession.id)).where(
                    AutoSession.status == "resolved"
                )
            )
        ).scalar() or 0
        handoff = (
            await self.db.execute(
                select(func.count(AutoSession.id)).where(
                    AutoSession.status == "handoff"
                )
            )
        ).scalar() or 0
        error = (
            await self.db.execute(
                select(func.count(AutoSession.id)).where(
                    AutoSession.status == "error"
                )
            )
        ).scalar() or 0
        auto_actions = (
            await self.db.execute(
                select(func.count(AutoAction.id)).where(
                    AutoAction.status == "success",
                    AutoAction.risk_level.in_(["read", "low"]),
                )
            )
        ).scalar() or 0
        approval_actions = (
            await self.db.execute(
                select(func.count(AutoAction.id)).where(
                    AutoAction.risk_level == "high"
                )
            )
        ).scalar() or 0

        by_scenario_rows = (
            await self.db.execute(
                select(AutoSession.scenario_key, func.count(AutoSession.id)).group_by(
                    AutoSession.scenario_key
                )
            )
        ).all()
        by_scenario = {k: v for k, v in by_scenario_rows if k}

        return {
            "total_sessions": total,
            "resolved_sessions": resolved,
            "handoff_sessions": handoff,
            "error_sessions": error,
            "auto_executed_actions": auto_actions,
            "approval_required_actions": approval_actions,
            "by_scenario": by_scenario,
        }


# --------------------------------------------------------------------------
# 后台编排入口
# --------------------------------------------------------------------------
async def run_session_in_background(session_id: str) -> None:
    """后台运行会话编排（独立 DB 会话，结束时提交/回滚）。"""
    factory = _get_session_factory()
    async with factory() as db:
        svc = AutoSessionService(db)
        try:
            await svc.start(session_id)
            await db.commit()
        except AutomationException as e:
            await db.rollback()
            logger.warning(f"编排业务异常 session={session_id}: {e.message}")
            # 标记转人工
            try:
                session = await svc.get_session(session_id)
                if session and session.status not in ("closed", "handoff"):
                    session.status = "handoff"
                    session.closed_by = "system(auto)"
                    await db.commit()
            except Exception:  # noqa: BLE001
                await db.rollback()
        except Exception as e:  # noqa: BLE001
            await db.rollback()
            logger.error(f"编排未预期异常 session={session_id}: {e}")
            try:
                session = await svc.get_session(session_id)
                if session and session.status not in ("closed", "handoff"):
                    session.status = "handoff"
                    session.closed_by = "system(auto)"
                    await db.commit()
            except Exception:  # noqa: BLE001
                await db.rollback()