LQAgentPlatform/views/debug/debug_api.py

"""
审查链路调试 API 端点

提供审查链路调试的 SSE 流式执行端点，以及完整的 Pydantic 请求/响应模型定义。

功能：
1. POST /debug/api/review/execute — 执行审查调试，SSE 流式返回进度
2. GET /debug/api/review/stream/{task_id} — SSE 断线重连
3. 并发控制（最大 5 个并发调试任务）
4. 全局超时控制（默认 180s）
5. 环节隔离模式支持
6. 调用记录自动保存
"""

import asyncio
import json
import logging
from datetime import datetime
from typing import Optional, List, Dict, AsyncGenerator, ClassVar

from fastapi import APIRouter, HTTPException, Path
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field, field_validator

from core.debug.sse_utils import (
    format_sse_event,
    sse_generator,
    debug_semaphore,
    _running_tasks,
    MAX_CONCURRENT_DEBUG_TASKS,
    DEBUG_GLOBAL_TIMEOUT,
    make_trace_id,
    make_record_id,
    CHAIN_NAMES,
    CHAIN_STEPS_COUNT,
)

logger = logging.getLogger(__name__)


# ============ 枚举常量 ============

class ChainId:
    """链路标识枚举"""
    COMPLETENESS = "completeness"
    TIMELINESS = "timeliness"
    REFERENCE = "reference"
    SENSITIVE = "sensitive"
    SEMANTIC = "semantic"
    GRAMMAR = "grammar"
    PROFESSIONAL = "professional"


class StepStatus:
    """步骤状态枚举"""
    PENDING = "pending"
    RUNNING = "running"
    SUCCESS = "success"
    ERROR = "error"


class RecordStatus:
    """记录状态枚举"""
    SUCC = "succ"
    FAIL = "fail"
    TIMEOUT = "timeout"


class ReviewType:
    """审查类型枚举"""
    BOTH = "both"
    NON_PARAMETER = "non_parameter"
    PARAMETER = "parameter"


# ============ 审查链路调试 - 请求/响应模型 ============

class RagParams(BaseModel):
    """专业性审查 RAG 参数"""
    review_type: str = Field(
        default="both",
        description="审查类型: both(全部) / non_parameter(非参数) / parameter(参数)",
    )
    collection_name: str = Field(
        default="construction_specs",
        description="Milvus 集合名称",
    )
    top_k: int = Field(
        default=5,
        ge=1,
        le=50,
        description="向量检索 top_k",
    )
    hybrid_top_k: int = Field(
        default=20,
        ge=1,
        le=100,
        description="混合检索 top_k",
    )
    dense_weight: float = Field(
        default=0.5,
        ge=0.0,
        le=1.0,
        description="稠密向量权重（0-1）",
    )
    parent_threshold: float = Field(
        default=0.3,
        ge=0.0,
        le=1.0,
        description="父文档增强阈值",
    )


class DebugExecuteRequest(BaseModel):
    """审查调试执行请求"""
    chain_id: str = Field(
        ...,
        description="链路标识: completeness / timeliness / reference / sensitive / semantic / grammar / professional",
    )
    content: str = Field(
        ...,
        description="待审查的方案内容",
    )
    reference: Optional[str] = Field(
        default=None,
        description="审查参考依据",
    )
    model: Optional[str] = Field(
        default=None,
        description="覆盖模型名称（如 deepseek_v3），与 function_name 互斥",
    )
    function_name: Optional[str] = Field(
        default=None,
        description="覆盖功能名称（如 completeness_review_generate），与 model 互斥",
    )
    prompt_version: Optional[str] = Field(
        default=None,
        description="指定提示词版本，null 则使用当前激活版本",
    )
    timeout: int = Field(
        default=60,
        ge=10,
        le=600,
        description="模型调用超时（秒）",
    )
    rag_params: Optional[RagParams] = Field(
        default=None,
        description="专业性审查专属参数，其他链路忽略",
    )
    isolation_mode: bool = Field(
        default=False,
        description="是否启用环节隔离模式",
    )
    isolation_steps: List[int] = Field(
        default_factory=list,
        description="隔离模式下要执行的步骤索引列表",
    )
    manual_inputs: Dict[str, str] = Field(
        default_factory=dict,
        description="隔离模式下各步骤的手动输入，key 为步骤索引，value 为输入内容",
    )

    _VALID_CHAIN_IDS: ClassVar[set] = {
        ChainId.COMPLETENESS,
        ChainId.TIMELINESS,
        ChainId.REFERENCE,
        ChainId.SENSITIVE,
        ChainId.SEMANTIC,
        ChainId.GRAMMAR,
        ChainId.PROFESSIONAL,
    }

    @field_validator("chain_id")
    @classmethod
    def check_chain_id(cls, v: str) -> str:
        """验证 chain_id 是否为合法枚举值"""
        if v not in cls._VALID_CHAIN_IDS:
            raise ValueError(
                f"chain_id 必须为以下值之一: "
                f"{json.dumps(sorted(cls._VALID_CHAIN_IDS), ensure_ascii=False)}"
            )
        return v

    @field_validator("content")
    @classmethod
    def check_content_not_empty(cls, v: str) -> str:
        """验证 content 不能为空"""
        if not v or not v.strip():
            raise ValueError("content 不能为空")
        return v


# ============ 提示词管理 - 请求/响应模型 ============

class PromptItem(BaseModel):
    """提示词列表项"""
    name: str = Field(default="", description="提示词名称")
    version: str = Field(default="", description="版本号")
    time: str = Field(default="", description="创建时间")
    chain: str = Field(default="", description="所属链路")
    is_current: bool = Field(default=False, description="是否为当前激活版本")
    note: str = Field(default="", description="版本说明")


class PromptListResponse(BaseModel):
    """提示词列表响应"""
    status: str = Field(default="not_implemented", description="状态")
    total: int = Field(default=0, description="总数")
    page: int = Field(default=1, description="当前页码")
    page_size: int = Field(default=50, description="每页条数")
    items: List[PromptItem] = Field(default_factory=list, description="提示词列表")
    chains: List[str] = Field(default_factory=list, description="可筛选的链路列表")


class PromptDetailResponse(BaseModel):
    """提示词详情响应"""
    status: str = Field(default="not_implemented", description="状态")
    name: str = Field(default="", description="提示词名称")
    version: str = Field(default="", description="版本号")
    time: str = Field(default="", description="创建时间")
    chain: str = Field(default="", description="所属链路")
    is_current: bool = Field(default=False, description="是否为当前激活版本")
    system_prompt: str = Field(default="", description="系统提示词")
    user_prompt: str = Field(default="", description="用户提示词模板")
    note: str = Field(default="", description="版本说明")
    variables: List[str] = Field(default_factory=list, description="模板变量列表")
    based_on: Optional[str] = Field(default=None, description="基于哪个版本")
    file_path: str = Field(default="", description="文件路径")


class PromptSaveRequest(BaseModel):
    """保存新版本请求"""
    name: str = Field(..., description="提示词名称")
    system_prompt: str = Field(..., description="系统提示词内容")
    user_prompt: str = Field(..., description="用户提示词模板")
    note: str = Field(default="", description="版本说明")
    set_current: bool = Field(default=True, description="是否设为当前激活版本")


class PromptSaveResponse(BaseModel):
    """保存新版本响应"""
    success: bool = Field(default=True, description="是否成功")
    name: str = Field(default="", description="提示词名称")
    version: str = Field(default="", description="新版本号")
    time: str = Field(default="", description="创建时间")
    message: str = Field(default="", description="消息说明")


class PromptCompareRequest(BaseModel):
    """版本对比请求"""
    name: str = Field(..., description="提示词名称")
    base_version: str = Field(..., description="基准版本号")
    target_version: str = Field(..., description="目标版本号")


class PromptCompareResponse(BaseModel):
    """版本对比响应"""
    status: str = Field(default="not_implemented", description="状态")
    name: str = Field(default="", description="提示词名称")
    base_version: str = Field(default="", description="基准版本")
    target_version: str = Field(default="", description="目标版本")
    diffs: List[dict] = Field(default_factory=list, description="差异列表")


class PromptActivateRequest(BaseModel):
    """激活版本请求"""
    name: str = Field(..., description="提示词名称")
    version: str = Field(..., description="版本号")


class PromptActivateResponse(BaseModel):
    """激活版本响应"""
    success: bool = Field(default=True, description="是否成功")
    name: str = Field(default="", description="提示词名称")
    version: str = Field(default="", description="版本号")
    message: str = Field(default="", description="消息说明")


class PromptVersionsResponse(BaseModel):
    """版本列表响应"""
    status: str = Field(default="not_implemented", description="状态")
    name: str = Field(default="", description="提示词名称")
    chain: str = Field(default="", description="所属链路")
    current_version: str = Field(default="", description="当前激活版本")
    versions: List[dict] = Field(default_factory=list, description="版本列表")


# ============ 调用记录 - 请求/响应模型 ============

class CallRecordItem(BaseModel):
    """调用记录列表项"""
    id: str = Field(default="", description="记录 ID")
    time: str = Field(default="", description="调用时间")
    chain: str = Field(default="", description="链路标识")
    chain_name: str = Field(default="", description="链路名称")
    doc_ref: str = Field(default="", description="文档引用")
    duration: str = Field(default="", description="持续时间（格式化）")
    duration_ms: int = Field(default=0, description="持续时间（毫秒）")
    status: str = Field(default="", description="状态: succ/fail/timeout")
    model: str = Field(default="", description="模型名称")
    prompt_ver: str = Field(default="", description="提示词版本")
    tokens: int = Field(default=0, description="Token 消耗")
    result_preview: str = Field(default="", description="结果预览")


class CallRecordListResponse(BaseModel):
    """调用记录列表响应"""
    status: str = Field(default="not_implemented", description="状态")
    total: int = Field(default=0, description="总数")
    page: int = Field(default=1, description="当前页码")
    page_size: int = Field(default=20, description="每页条数")
    total_pages: int = Field(default=0, description="总页数")
    items: List[CallRecordItem] = Field(default_factory=list, description="记录列表")
    chains: List[str] = Field(default_factory=list, description="可筛选的链路列表")
    status_counts: Dict[str, int] = Field(default_factory=dict, description="状态统计")


class StepDetail(BaseModel):
    """步骤详情"""
    index: int = Field(default=0, description="步骤索引")
    name: str = Field(default="", description="步骤名称")
    duration_ms: int = Field(default=0, description="持续时间（毫秒）")
    status: str = Field(default="", description="状态: succ/fail/timeout")
    input: dict = Field(default_factory=dict, description="步骤输入")
    output: dict = Field(default_factory=dict, description="步骤输出")


class CallRecordDetailResponse(BaseModel):
    """调用记录详情响应"""
    id: str = Field(default="", description="记录 ID")
    time: str = Field(default="", description="调用时间")
    chain: str = Field(default="", description="链路标识")
    chain_name: str = Field(default="", description="链路名称")
    doc_ref: str = Field(default="", description="文档引用")
    status: str = Field(default="", description="状态: succ/fail/timeout")
    duration_ms: int = Field(default=0, description="持续时间（毫秒）")
    model: str = Field(default="", description="模型名称")
    function_name: Optional[str] = Field(default=None, description="功能名称")
    prompt_ver: str = Field(default="", description="提示词版本")
    prompt_name: str = Field(default="", description="提示词名称")
    tokens: int = Field(default=0, description="Token 消耗")
    params: dict = Field(default_factory=dict, description="请求参数")
    steps: List[StepDetail] = Field(default_factory=list, description="步骤列表")
    result: str = Field(default="", description="审查结果")
    error_message: Optional[str] = Field(default=None, description="错误信息")


class OverrideParams(BaseModel):
    """回放覆盖参数"""
    model: Optional[str] = Field(default=None, description="覆盖模型名称")
    prompt_version: Optional[str] = Field(default=None, description="覆盖提示词版本")
    rag_params: Optional[RagParams] = Field(default=None, description="覆盖 RAG 参数")


class ReplayRequest(BaseModel):
    """回放调用请求"""
    record_id: str = Field(..., description="要回放的调用记录 ID")
    override_params: Optional[OverrideParams] = Field(
        default=None,
        description="覆盖原始参数，不指定则使用原始参数",
    )


class ExportRequest(BaseModel):
    """导出调用记录请求"""
    record_ids: List[str] = Field(..., description="要导出的记录 ID 列表")
    format: str = Field(default="json", description="导出格式")


# ============ 路由定义 ============

debug_router = APIRouter(prefix="/debug", tags=["审查调试工作台"])

# 注册子模块路由
from .prompt_api import register_routes as register_prompt_routes
from .record_api import register_routes as register_record_routes
from .rag_debug_api import register_routes as register_rag_debug_routes

register_prompt_routes(debug_router)
register_record_routes(debug_router)
register_rag_debug_routes(debug_router)


# ============ 审查链路调试端点 ============


@debug_router.post("/api/review/execute")
async def execute_review(request: DebugExecuteRequest):
    """
    执行审查调试（SSE 流式返回）

    支持全部 7 个审查链路：
    - completeness: 完整性审查
    - timeliness: 时效性审查
    - reference: 规范性审查
    - sensitive: 敏感词检查
    - semantic: 语义逻辑检查
    - grammar: 语法检查
    - professional: 专业性审查

    支持环节隔离模式，可单独执行指定步骤。
    调试执行不进 Celery 队列，直接在请求协程中执行，支持实时 SSE 推送。

    并发限制：最多 5 个调试任务同时执行。
    超时控制：全局默认 180s。
    """
    # ---- 检查并发上限 ----
    if debug_semaphore.locked():
        raise HTTPException(
            status_code=429,
            detail=(
                f"并发调试任务数已达上限 ({MAX_CONCURRENT_DEBUG_TASKS})，"
                f"请等待其他任务完成后再试"
            ),
        )

    chain_id = request.chain_id
    total_steps = CHAIN_STEPS_COUNT.get(chain_id, 3)

    async def event_generator():
        async with debug_semaphore:
            event_queue: asyncio.Queue = asyncio.Queue()
            task_id = make_trace_id(chain_id)
            record_id = make_record_id()

            # 缓存任务队列供断线重连
            _running_tasks[task_id] = event_queue

            # ---- 发送 started 事件 ----
            yield format_sse_event("started", {
                "task_id": task_id,
                "chain_id": chain_id,
                "total_steps": total_steps,
            })

            # ---- 后台执行任务 ----
            exec_task = asyncio.create_task(
                _run_debug_execution(request, event_queue, task_id, record_id)
            )

            completed_data = None
            error_occurred = None
            steps_collected: List[dict] = []

            try:
                # ---- 消费队列事件 ----
                while True:
                    try:
                        event_type, data = await asyncio.wait_for(
                            event_queue.get(),
                            timeout=DEBUG_GLOBAL_TIMEOUT,
                        )
                    except asyncio.TimeoutError:
                        error_occurred = "执行超时"
                        yield format_sse_event("error", {
                            "task_id": task_id,
                            "message": f"执行超时 (>{DEBUG_GLOBAL_TIMEOUT}s)",
                        })
                        exec_task.cancel()
                        break

                    if event_type == "__end__":
                        break

                    # 跳过 executor 的 started（我们已发送自定义 started）
                    if event_type == "started":
                        continue

                    # 收集步骤数据用于保存记录
                    if event_type == "step_result":
                        steps_collected.append({
                            "index": data.get("step_index"),
                            "name": data.get("step_name"),
                            "status": data.get("status"),
                            "duration_ms": int((data.get("duration") or 0) * 1000),
                            "input": data.get("input", {}),
                            "output": data.get("output", {}),
                        })

                    if event_type == "error":
                        error_occurred = data.get("message", "未知错误")

                    if event_type == "completed":
                        completed_data = data

                    yield format_sse_event(event_type, data)

                    if event_type == "completed":
                        break

            except asyncio.CancelledError:
                yield format_sse_event("error", {
                    "task_id": task_id,
                    "message": "连接已断开，任务被取消",
                })
            except Exception as exc:
                logger.exception("[execute_review] 事件处理异常")
                yield format_sse_event("error", {
                    "task_id": task_id,
                    "message": str(exc),
                })
            finally:
                _running_tasks.pop(task_id, None)
                if not exec_task.done():
                    exec_task.cancel()

                # ---- 自动保存调用记录 ----
                try:
                    await _save_debug_record(
                        request=request,
                        task_id=task_id,
                        record_id=record_id,
                        chain_id=chain_id,
                        completed_data=completed_data,
                        error_message=error_occurred,
                        steps=steps_collected,
                    )
                except Exception as exc:
                    logger.warning("[execute_review] 保存调用记录失败: %s", exc)

    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",
        },
    )


# ============ SSE 断线重连端点 ============


@debug_router.get("/api/review/stream/{task_id}")
async def stream_review_progress(task_id: str = Path(..., description="任务 ID")):
    """
    重新连接获取审查调试进度（SSE 流式）

    当 SSE 连接断开时，用于重新连接获取仍在执行中的任务进度。
    从当前进度继续推送，不重放已完成的 step 事件。
    """
    async def event_generator():
        queue = _running_tasks.get(task_id)
        if queue is None:
            yield format_sse_event("error", {
                "task_id": task_id,
                "message": "任务不存在或已完成",
            })
            return

        # 发送 resumed 标记
        yield format_sse_event("started", {
            "task_id": task_id,
            "resumed": True,
        })

        # 继续从原队列消费事件
        while True:
            try:
                event_type, data = await asyncio.wait_for(
                    queue.get(),
                    timeout=DEBUG_GLOBAL_TIMEOUT,
                )
            except asyncio.TimeoutError:
                yield format_sse_event("error", {
                    "task_id": task_id,
                    "message": "重连等待超时",
                })
                break

            if event_type == "__end__":
                break
            yield format_sse_event(event_type, data)
            if event_type == "completed":
                break
            queue.task_done()

    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",
        },
    )


# ============================================================
# 内部辅助函数
# ============================================================


async def _run_debug_execution(
    request: DebugExecuteRequest,
    event_queue: asyncio.Queue,
    task_id: str,
    record_id: str,
) -> None:
    """
    执行审查调试后台任务。

    根据 isolation_mode 选择执行路径：
    - True: 使用 IsolationRunner.run_selected_steps()
    - False: 使用 DebugExecutor.execute()
    """
    try:
        if request.isolation_mode:
            await _run_isolation_mode(request, event_queue, task_id)
        else:
            await _run_normal_mode(request, event_queue, task_id, record_id)
    except asyncio.CancelledError:
        await event_queue.put(("error", {
            "task_id": task_id,
            "message": "执行被取消",
        }))
    except Exception as exc:
        logger.exception("[_run_debug_execution] 执行异常")
        await event_queue.put(("error", {
            "task_id": task_id,
            "message": str(exc),
        }))


async def _run_normal_mode(
    request: DebugExecuteRequest,
    event_queue: asyncio.Queue,
    task_id: str,
    record_id: str,
) -> None:
    """正常模式：使用 DebugExecutor 执行"""
    from core.debug.executor import DebugExecutor

    executor = DebugExecutor()
    await executor.execute(request, event_queue)


async def _run_isolation_mode(
    request: DebugExecuteRequest,
    event_queue: asyncio.Queue,
    task_id: str,
) -> None:
    """隔离模式：使用 IsolationRunner 执行选中的步骤"""
    from core.debug.isolation_runner import IsolationRunner
    from core.debug.step_dispatcher import CHAIN_STEPS

    runner = IsolationRunner()
    chain_id = request.chain_id

    params = {
        "content": request.content,
        "reference": request.reference or "",
        "model": request.model,
        "function_name": request.function_name,
        "timeout": request.timeout,
    }
    if request.rag_params:
        params["rag_params"] = request.rag_params

    manual_inputs = request.manual_inputs or {}
    selected_indices = list(request.isolation_steps) if request.isolation_steps else []

    step_result_dicts = await runner.run_selected_steps(
        chain_id=chain_id,
        selected_indices=selected_indices,
        manual_inputs=manual_inputs,
        **params,
    )

    # 从 step_def 获取 phase 信息
    step_defs = CHAIN_STEPS.get(chain_id, [])
    current_phase = None

    for sr in step_result_dicts:
        si = sr.get("index", 0)

        # 获取 phase 信息
        phase = None
        if si < len(step_defs):
            sd = step_defs[si]
            if hasattr(sd, "phase"):
                phase = sd.phase
            elif isinstance(sd, dict):
                phase = sd.get("phase")

        # phase_label：阶段切换时推送
        if phase and phase != current_phase:
            current_phase = phase
            await event_queue.put(("phase_label", {
                "task_id": task_id,
                "label": phase,
            }))

        # step_progress
        await event_queue.put(("step_progress", {
            "task_id": task_id,
            "step_index": si,
            "step_name": sr.get("name", ""),
            "status": "running",
            "duration": None,
        }))

        # step_result
        await event_queue.put(("step_result", {
            "task_id": task_id,
            "step_index": si,
            "step_name": sr.get("name", ""),
            "status": sr.get("status", ""),
            "duration": sr.get("duration", 0),
            "input": sr.get("input", {}),
            "output": sr.get("output", {}),
            "error": sr.get("error"),
        }))

    # completed 事件
    total_duration = sum(
        sr.get("duration", 0) or 0 for sr in step_result_dicts
    )
    success_count = sum(
        1 for sr in step_result_dicts if sr.get("status") == "success"
    )
    error_count = sum(
        1 for sr in step_result_dicts if sr.get("status") == "error"
    )
    skipped_count = sum(
        1 for sr in step_result_dicts if sr.get("status") == "skipped"
    )

    await event_queue.put(("completed", {
        "task_id": task_id,
        "chain_id": chain_id,
        "total_duration": round(total_duration, 3),
        "record_id": "",
        "final_result": {
            "summary": (
                f"{success_count}/{len(step_result_dicts)} 步骤成功, "
                f"{error_count} 错误, {skipped_count} 跳过"
            ),
            "success_count": success_count,
            "error_count": error_count,
            "skipped_count": skipped_count,
            "total_steps": len(step_result_dicts),
        },
    }))


async def _save_debug_record(
    request: DebugExecuteRequest,
    task_id: str,
    record_id: str,
    chain_id: str,
    completed_data: Optional[dict],
    error_message: Optional[str],
    steps: List[dict],
) -> None:
    """通过 RecordManager 保存调用记录"""
    from core.debug.record_manager import RecordManager

    duration_ms = 0
    final_result = ""
    if completed_data:
        duration_sec = completed_data.get("total_duration", 0) or 0
        duration_ms = int(duration_sec * 1000)
        final_result = str(completed_data.get("final_result", {}).get("summary", ""))

    status = "succ"
    if error_message:
        if "超时" in str(error_message):
            status = "timeout"
        else:
            status = "fail"

    record_data = {
        "id": record_id,
        "time": datetime.now().isoformat(),
        "chain": chain_id,
        "chain_name": CHAIN_NAMES.get(chain_id, ""),
        "doc_ref": "",
        "status": status,
        "duration_ms": duration_ms,
        "model": request.model or "default",
        "function_name": request.function_name or "",
        "prompt_ver": request.prompt_version or "",
        "prompt_name": "",
        "tokens": 0,
        "params": {
            "review_content": request.content,
            "review_references": request.reference or "",
            "model_override": request.model,
            "function_name": request.function_name,
            "timeout": request.timeout,
        },
        "execution_params": {
            "isolation_mode": request.isolation_mode,
            "isolation_steps": list(request.isolation_steps),
            "rag_params": (
                request.rag_params.model_dump()
                if request.rag_params else None
            ),
        },
        "steps": steps,
        "result": final_result,
        "error_message": error_message,
    }

    rm = RecordManager()
    await rm.save_record(record_data)