Neo-ZQYY/docs/specs/ai-monitoring-testing/design.md

# 设计文档 — P15：AI 监控后台 + 测试重建 + 回填

> 权威参考：**实施过程中如遇细节不明确，应优先查阅 PRD 原文 `docs/prd/specs/P15-ai-monitoring-testing.md`**

## 概述

本设计覆盖六个模块：
- **模块 A**：后端 API（13 个端点，`/api/admin/ai/*`）
- **模块 B**：前端页面（4 个 admin-web 页面：Dashboard、调度状态、调用明细、手动操作）
- **模块 C**：测试体系重建（旧脚本归档 + 15 场景全链路测试 + 7 个属性测试更新）
- **模块 D**：历史数据回填（`scripts/ops/ai_backfill.py`，分批 + 断点续跑）
- **模块 E**：数据保留与清理（定时任务，每日 03:00）
- **模块 F**：查询优化与文档收尾（BRIN 索引 + 文档同步）

### 设计决策与理由

| 决策 | 理由 |
|------|------|
| 告警状态用 `ai_run_logs.alert_status` 字段而非独立告警表 | 告警来源就是 `ai_run_logs` 中的失败/超时/熔断记录，独立表需要同步维护且增加 JOIN 开销；单字段方案查询简单、无数据一致性风险 |
| `alert_status` 默认 `NULL`（非告警记录）而非 `pending` | 绝大多数记录是成功的，NULL 避免无意义的存储和索引膨胀；仅 status IN ('failed','timeout','circuit_open') 的记录才设置 `alert_status='pending'` |
| 批量执行用 `batch_id` + 内存缓存而非持久化表 | 批量操作是低频管理操作，batch_id 有效期短（10 分钟），无需持久化；内存 dict 足够，单实例部署无分布式问题 |
| 清理定时任务复用现有 `Scheduler` + `scheduled_tasks` | 后端已有调度器基础设施（P14），新增一条 `scheduled_tasks` 记录即可，无需引入额外定时框架 |
| BRIN 索引而非 B-tree 索引用于 `ai_run_logs.created_at` | `ai_run_logs` 按时间顺序插入，BRIN 索引体积小（约 B-tree 的 1/100），适合时间范围扫描的 Dashboard 聚合查询 |
| 全链路测试 Mock + 真实 API 双轨 | Mock 模式保证 CI 稳定性和速度；真实 API 模式用于集成验证，通过环境变量 `AI_TEST_MODE` 切换 |
| 回填脚本独立运行而非通过 admin-web 触发 | 回填是一次性操作，脚本可断点续跑、可在终端监控进度；admin-web 的 batch-run API 用于日常小批量操作 |
| 前端 4 个页面各自独立文件而非 Tab 切换 | 每个页面功能独立、数据源不同，独立路由便于直接链接和浏览器历史导航 |


## 架构

### 整体架构

```mermaid
graph TB
    subgraph "admin-web 前端"
        DASH["Dashboard 页面<br/>统计卡片+趋势图+饼图+预算+告警"]
        SCHED["调度状态页面<br/>分页表格+筛选+去重统计"]
        LOGS["调用明细页面<br/>分页表格+详情抽屉"]
        OPS["手动操作页面<br/>重跑+缓存失效+批量+告警"]
    end

    subgraph "后端 API 层"
        ROUTER["admin_ai.py<br/>13 个端点<br/>/api/admin/ai/*"]
    end

    subgraph "服务层"
        ADMIN_SVC["AdminAIService<br/>聚合查询+批量操作+告警管理"]
    end

    subgraph "P14 已有基础设施（复用）"
        DISP["AIDispatcher<br/>事件调度+调用链"]
        CB["CircuitBreaker"]
        RL["RateLimiter"]
        BT["BudgetTracker"]
        LOG_SVC["AIRunLogService"]
        CACHE_SVC["AICacheService"]
    end

    subgraph "存储层"
        DB_LOG["ai_run_logs<br/>+alert_status 字段"]
        DB_JOB["ai_trigger_jobs"]
        DB_CACHE["ai_cache"]
        DB_CLUE["member_retention_clue"]
    end

    subgraph "定时任务"
        CLEANUP["数据清理任务<br/>每日 03:00"]
    end

    subgraph "运维脚本"
        BACKFILL["ai_backfill.py<br/>历史数据回填"]
    end

    DASH --> ROUTER
    SCHED --> ROUTER
    LOGS --> ROUTER
    OPS --> ROUTER

    ROUTER --> ADMIN_SVC
    ADMIN_SVC --> LOG_SVC
    ADMIN_SVC --> CACHE_SVC
    ADMIN_SVC --> DISP
    ADMIN_SVC --> BT

    ADMIN_SVC --> DB_LOG
    ADMIN_SVC --> DB_JOB
    ADMIN_SVC --> DB_CACHE

    CLEANUP --> DB_LOG
    CLEANUP --> DB_JOB
    CLEANUP --> DB_CACHE

    BACKFILL --> DISP
    BACKFILL --> DB_CLUE
```

### 成本二次确认流程

```mermaid
sequenceDiagram
    participant Admin as 管理员
    participant FE as admin-web
    participant API as /api/admin/ai
    participant SVC as AdminAIService
    participant MEM as 内存 batch_store
    participant DISP as AIDispatcher

    Admin->>FE: 选择 App + 会员 + 门店
    FE->>API: POST /batch-run {app_types, member_ids, site_id}
    API->>SVC: estimate_batch()
    SVC-->>API: {batch_id, estimated_calls, estimated_tokens}
    API-->>FE: 返回预估信息
    SVC->>MEM: 存储 batch_id → params（TTL 10min）

    FE->>Admin: 弹窗："将执行 N 次调用，预估 M tokens"
    Admin->>FE: 点击"确认执行"
    FE->>API: POST /batch-run/confirm {batch_id}
    API->>SVC: confirm_batch(batch_id)
    SVC->>MEM: 取出并删除 batch_id
    SVC->>DISP: asyncio.create_task(execute_batch)
    API-->>FE: {status: "started"}
```

### 数据清理流程

```mermaid
flowchart TD
    A[定时任务触发<br/>每日 03:00] --> B[删除 90 天前<br/>ai_run_logs]
    B --> C[删除 90 天前<br/>ai_trigger_jobs]
    C --> D{遍历每个 App 类型<br/>App2~App8}
    D --> E[统计该 App 的<br/>ai_cache 记录数]
    E --> F{记录数 > 20000?}
    F -->|是| G[删除最旧的超出记录]
    F -->|否| H[跳过]
    G --> D
    H --> D
    D -->|全部完成| I[记录清理结果到日志]
```


## 组件与接口

### 模块 A：后端 API

#### 1. admin_ai.py — AI 监控路由（新建）

文件：`apps/backend/app/routers/admin_ai.py`

```python
router = APIRouter(prefix="/api/admin/ai", tags=["admin-ai"])

# ---- Dashboard ----
@router.get("/dashboard")
async def get_dashboard(site_id: int | None = None, user=Depends(require_admin)) -> DashboardResponse:
    """总览统计：今日调用/成功率/Token/延迟 + 7天趋势 + App分布 + 预算 + 告警 + App健康。"""

# ---- 调度任务 ----
@router.get("/trigger-jobs")
async def list_trigger_jobs(
    event_type: str | None = None, status: str | None = None,
    site_id: int | None = None, date_from: str | None = None, date_to: str | None = None,
    page: int = 1, page_size: int = 20, user=Depends(require_admin),
) -> TriggerJobListResponse:
    """调度任务分页列表 + 今日去重统计。"""

@router.get("/trigger-jobs/{job_id}")
async def get_trigger_job(job_id: int, user=Depends(require_admin)) -> TriggerJobDetailResponse:
    """调度任务详情（含 payload、error_message）。"""

@router.post("/trigger-jobs/{job_id}/retry")
async def retry_trigger_job(job_id: int, user=Depends(require_admin)) -> RetryResponse:
    """手动重跑：创建新 trigger_job（is_forced=true），异步执行。"""

# ---- 调用记录 ----
@router.get("/run-logs")
async def list_run_logs(
    app_type: str | None = None, status: str | None = None, trigger_type: str | None = None,
    site_id: int | None = None, date_from: str | None = None, date_to: str | None = None,
    page: int = 1, page_size: int = 20, user=Depends(require_admin),
) -> RunLogListResponse:
    """调用记录分页列表。"""

@router.get("/run-logs/{log_id}")
async def get_run_log(log_id: int, user=Depends(require_admin)) -> RunLogDetailResponse:
    """调用记录详情（含完整 prompt/response/error，不脱敏）。"""

# ---- 缓存管理 ----
@router.post("/cache/invalidate")
async def invalidate_cache(body: CacheInvalidateRequest, user=Depends(require_admin)) -> CacheInvalidateResponse:
    """批量缓存失效：将匹配条件的 ai_cache.status 设为 invalidated。"""

# ---- Token 预算 ----
@router.get("/budget")
async def get_budget(user=Depends(require_admin)) -> BudgetResponse:
    """Token 预算使用情况：日/月已用量、上限、百分比。"""

# ---- 批量执行 ----
@router.post("/batch-run")
async def create_batch_run(body: BatchRunRequest, user=Depends(require_admin)) -> BatchRunEstimate:
    """创建批量执行请求，返回预估（不立即执行）。"""

@router.post("/batch-run/confirm")
async def confirm_batch_run(body: BatchRunConfirm, user=Depends(require_admin)) -> BatchRunConfirmResponse:
    """确认批量执行，后台异步执行。"""

# ---- 告警管理 ----
@router.get("/alerts")
async def list_alerts(
    alert_status: str | None = None, site_id: int | None = None,
    page: int = 1, page_size: int = 20, user=Depends(require_admin),
) -> AlertListResponse:
    """告警列表（ai_run_logs WHERE status IN ('failed','timeout','circuit_open')）。"""

@router.post("/alerts/{log_id}/ack")
async def ack_alert(log_id: int, user=Depends(require_admin)) -> AlertActionResponse:
    """确认告警：alert_status → acknowledged。"""

@router.post("/alerts/{log_id}/ignore")
async def ignore_alert(log_id: int, user=Depends(require_admin)) -> AlertActionResponse:
    """忽略告警：alert_status → ignored。"""
```

#### 2. admin_ai Schema（新建）

文件：`apps/backend/app/schemas/admin_ai.py`

```python
# ---- Dashboard ----
class DashboardResponse(BaseModel):
    today_calls: int
    today_success_rate: float          # 0.0 ~ 1.0
    today_tokens: int
    today_avg_latency_ms: float
    trend_7d: list[DailyTrend]         # 近 7 天按日聚合
    app_distribution: list[AppDistItem] # 各 App 调用占比
    budget: BudgetInfo                  # 日/月预算进度
    recent_alerts: list[AlertItem]      # 最近告警
    app_health: list[AppHealthItem]     # 各 App 最近调用状态

class DailyTrend(BaseModel):
    date: str                           # YYYY-MM-DD
    calls: int
    success_rate: float

class AppDistItem(BaseModel):
    app_type: str
    count: int
    percentage: float

class BudgetInfo(BaseModel):
    daily_used: int
    daily_limit: int
    daily_pct: float
    monthly_used: int
    monthly_limit: int
    monthly_pct: float

class AlertItem(BaseModel):
    id: int
    app_type: str
    status: str                         # failed / timeout / circuit_open
    alert_status: str | None            # pending / acknowledged / ignored
    error_message: str | None
    created_at: str

class AppHealthItem(BaseModel):
    app_type: str
    last_status: str | None
    last_call_at: str | None

# ---- 调度任务 ----
class TriggerJobListResponse(BaseModel):
    items: list[TriggerJobItem]
    total: int
    page: int
    page_size: int
    today_skipped_duplicates: int       # 今日去重跳过数

class TriggerJobItem(BaseModel):
    id: int
    event_type: str
    member_id: int | None
    status: str
    app_chain: str | None
    is_forced: bool
    site_id: int
    started_at: str | None
    finished_at: str | None
    created_at: str

class TriggerJobDetailResponse(TriggerJobItem):
    payload: dict | None
    error_message: str | None
    connector_type: str

class RetryResponse(BaseModel):
    trigger_job_id: int
    status: str                         # "pending"

# ---- 调用记录 ----
class RunLogListResponse(BaseModel):
    items: list[RunLogItem]
    total: int
    page: int
    page_size: int

class RunLogItem(BaseModel):
    id: int
    app_type: str
    trigger_type: str
    member_id: int | None
    tokens_used: int
    latency_ms: int | None
    status: str
    site_id: int
    created_at: str

class RunLogDetailResponse(RunLogItem):
    request_prompt: str | None          # 完整 prompt，不截断
    response_text: str | None           # 完整 response
    error_message: str | None
    session_id: str | None
    finished_at: str | None

# ---- 缓存失效 ----
class CacheInvalidateRequest(BaseModel):
    site_id: int                        # 必填
    app_type: str | None = None         # 可选
    member_id: int | None = None        # 可选

class CacheInvalidateResponse(BaseModel):
    affected_count: int

# ---- Token 预算 ----
class BudgetResponse(BaseModel):
    daily_used: int
    daily_limit: int
    daily_pct: float
    monthly_used: int
    monthly_limit: int
    monthly_pct: float

# ---- 批量执行 ----
class BatchRunRequest(BaseModel):
    app_types: list[str]
    member_ids: list[int]
    site_id: int

class BatchRunEstimate(BaseModel):
    batch_id: str
    estimated_calls: int
    estimated_tokens: int

class BatchRunConfirm(BaseModel):
    batch_id: str

class BatchRunConfirmResponse(BaseModel):
    status: str                         # "started"

# ---- 告警 ----
class AlertListResponse(BaseModel):
    items: list[AlertItem]
    total: int
    page: int
    page_size: int

class AlertActionResponse(BaseModel):
    id: int
    alert_status: str
```

#### 3. AdminAIService（新建）

文件：`apps/backend/app/services/ai/admin_service.py`

```python
class AdminAIService:
    """AI 监控后台聚合服务。"""

    def __init__(self, dispatcher: AIDispatcher, budget_tracker: BudgetTracker):
        self._dispatcher = dispatcher
        self._budget = budget_tracker
        self._batch_store: dict[str, BatchInfo] = {}  # batch_id → params + 过期时间

    # ---- Dashboard ----
    async def get_dashboard(self, site_id: int | None = None) -> dict:
        """聚合 Dashboard 所有数据：今日统计、7天趋势、App分布、预算、告警、App健康。"""

    async def _get_today_stats(self, site_id: int | None) -> dict:
        """今日调用次数、成功率、Token 消耗、平均延迟。"""

    async def _get_7d_trend(self, site_id: int | None) -> list[dict]:
        """近 7 天按日聚合。"""

    async def _get_app_distribution(self, site_id: int | None) -> list[dict]:
        """各 App 调用占比。"""

    async def _get_app_health(self, site_id: int | None) -> list[dict]:
        """各 App 最近一次调用状态。"""

    # ---- 调度任务 ----
    async def list_trigger_jobs(self, filters: dict, page: int, page_size: int) -> dict:
        """分页查询 ai_trigger_jobs + 今日去重统计。"""

    async def get_trigger_job(self, job_id: int) -> dict:
        """单条调度任务详情。"""

    async def retry_trigger_job(self, job_id: int) -> int:
        """手动重跑：创建新 trigger_job（is_forced=true），调用 dispatcher 异步执行。返回新 job_id。"""

    # ---- 调用记录 ----
    async def list_run_logs(self, filters: dict, page: int, page_size: int) -> dict:
        """分页查询 ai_run_logs。"""

    async def get_run_log(self, log_id: int) -> dict:
        """单条调用记录详情（含完整 prompt/response）。"""

    # ---- 缓存管理 ----
    async def invalidate_cache(self, site_id: int, app_type: str | None, member_id: int | None) -> int:
        """批量缓存失效，返回受影响记录数。"""

    # ---- Token 预算 ----
    async def get_budget(self) -> dict:
        """Token 预算使用情况。"""

    # ---- 批量执行 ----
    async def estimate_batch(self, app_types: list[str], member_ids: list[int], site_id: int) -> dict:
        """计算预估，生成 batch_id 存入内存（TTL 10min）。"""

    async def confirm_batch(self, batch_id: str) -> None:
        """确认批量执行，取出 batch_id 参数，异步执行。"""

    def _cleanup_expired_batches(self) -> None:
        """清理过期的 batch_id。"""

    # ---- 告警管理 ----
    async def list_alerts(self, alert_status: str | None, site_id: int | None, page: int, page_size: int) -> dict:
        """告警列表：ai_run_logs WHERE status IN ('failed','timeout','circuit_open')。"""

    async def ack_alert(self, log_id: int) -> str:
        """确认告警：UPDATE ai_run_logs SET alert_status='acknowledged'。"""

    async def ignore_alert(self, log_id: int) -> str:
        """忽略告警：UPDATE ai_run_logs SET alert_status='ignored'。"""
```

#### 4. 数据清理服务（新建）

文件：`apps/backend/app/services/ai/cleanup_service.py`

```python
class AICleanupService:
    """AI 数据清理服务，由定时任务调用。"""

    RETENTION_DAYS = 90
    CACHE_LIMIT_PER_APP = 20_000
    CACHE_APP_TYPES = [
        "app2_finance", "app3_clue", "app4_analysis", "app5_tactics",
        "app6_note_analysis", "app7_customer_analysis", "app8_clue_consolidated",
    ]

    async def run_cleanup(self) -> dict:
        """执行全部清理，返回各步骤删除记录数。"""

    async def _cleanup_run_logs(self) -> int:
        """DELETE FROM ai_run_logs WHERE created_at < now() - 90 days。"""

    async def _cleanup_trigger_jobs(self) -> int:
        """DELETE FROM ai_trigger_jobs WHERE created_at < now() - 90 days。"""

    async def _cleanup_cache(self) -> dict[str, int]:
        """每个 App 类型保留最新 20,000 条，删除超出部分。"""
```

### 模块 B：前端页面

#### 5. admin-web API 层（新建）

文件：`apps/admin-web/src/api/adminAI.ts`

```typescript
// Dashboard
export const getDashboard = (siteId?: number) =>
  client.get('/api/admin/ai/dashboard', { params: { site_id: siteId } });

// 调度任务
export const getTriggerJobs = (params: TriggerJobQuery) =>
  client.get('/api/admin/ai/trigger-jobs', { params });
export const getTriggerJobDetail = (id: number) =>
  client.get(`/api/admin/ai/trigger-jobs/${id}`);
export const retryTriggerJob = (id: number) =>
  client.post(`/api/admin/ai/trigger-jobs/${id}/retry`);

// 调用记录
export const getRunLogs = (params: RunLogQuery) =>
  client.get('/api/admin/ai/run-logs', { params });
export const getRunLogDetail = (id: number) =>
  client.get(`/api/admin/ai/run-logs/${id}`);

// 缓存管理
export const invalidateCache = (data: CacheInvalidateReq) =>
  client.post('/api/admin/ai/cache/invalidate', data);

// Token 预算
export const getBudget = () =>
  client.get('/api/admin/ai/budget');

// 批量执行
export const createBatchRun = (data: BatchRunReq) =>
  client.post('/api/admin/ai/batch-run', data);
export const confirmBatchRun = (batchId: string) =>
  client.post('/api/admin/ai/batch-run/confirm', { batch_id: batchId });

// 告警
export const getAlerts = (params: AlertQuery) =>
  client.get('/api/admin/ai/alerts', { params });
export const ackAlert = (id: number) =>
  client.post(`/api/admin/ai/alerts/${id}/ack`);
export const ignoreAlert = (id: number) =>
  client.post(`/api/admin/ai/alerts/${id}/ignore`);
```

#### 6. 前端页面组件

文件：`apps/admin-web/src/pages/AIDashboard.tsx`（新建）

Dashboard 页面布局：
- 顶部：门店筛选 Select + 刷新按钮
- 第一行：4 个 Statistic 卡片（今日调用、成功率、Token 消耗、平均延迟）
- 第二行左：7 天趋势折线图（Ant Design Charts / 内联 SVG）
- 第二行右：App 调用占比饼图
- 第三行左：Token 预算进度条（日/月各一个 Progress）
- 第三行右：App 健康状态列表（Badge 标记状态）
- 第四行：告警列表 Table + App 配置信息（只读 Descriptions）

文件：`apps/admin-web/src/pages/AITriggerJobs.tsx`（新建）

调度状态页面布局：
- 顶部：筛选器行（event_type Select + status Select + site_id Select + DatePicker.RangePicker）
- 统计行：今日去重跳过数 Statistic
- 主体：Table 分页表格（事件类型、会员、状态 Tag、执行链、耗时、操作列）
- 操作列：查看详情 → Modal、手动重跑 → Popconfirm

文件：`apps/admin-web/src/pages/AIRunLogs.tsx`（新建）

调用明细页面布局：
- 顶部：筛选器行（app_type + status + trigger_type + site_id + DatePicker.RangePicker）
- 主体：Table 分页表格（app_type、trigger_type、member_id、tokens、延迟、状态 Tag）
- 点击行：Drawer 抽屉展示完整 prompt（代码块）、完整 response（代码块）、error_message

文件：`apps/admin-web/src/pages/AIOperations.tsx`（新建）

手动操作页面布局（4 个 Card 区域）：
- Card 1 — 手动重跑：App Select + member_id Input + site_id Select + 执行 Button
- Card 2 — 缓存失效：app_type Select + member_id Input + site_id Select + 失效 Button + 结果 message
- Card 3 — 批量执行：app_types Checkbox.Group + member_ids TextArea + site_id Select → 预估 → 确认弹窗 Modal
- Card 4 — 告警管理：Table（告警列表）+ 每行"确认"/"忽略" Button

### 模块 D：回填脚本

#### 7. ai_backfill.py（新建）

文件：`scripts/ops/ai_backfill.py`

```python
"""历史数据回填脚本。

用法：
    cd C:\\Project\\NeoZQYY
    uv run python scripts/ops/ai_backfill.py [--dry-run] [--batch-size 10] [--interval 5]

功能：
    1. 查询半年内活跃会员（消费/备注/任务变更三者并集）
    2. 分批执行 AI 调用链（每批 10 人，间隔 5 秒）
    3. 断点续跑（已完成 member_id 记录到本地文件）
    4. App8 写 member_retention_clue（DELETE + INSERT 事务）
"""

SITE_ID = 2790685415443269
CHECKPOINT_FILE = "scripts/ops/_ai_backfill_checkpoint.json"
BATCH_SIZE = 10
BATCH_INTERVAL = 5  # 秒
DATE_FROM = "2025-09-21"
DATE_TO = "2026-03-21"

class AIBackfillRunner:
    def __init__(self, dry_run: bool = False):
        self.dry_run = dry_run
        self.completed: set[int] = set()

    def load_checkpoint(self) -> None:
        """从本地文件加载已完成的 member_id 集合。"""

    def save_checkpoint(self) -> None:
        """保存已完成的 member_id 到本地文件。"""

    async def query_active_members(self) -> list[int]:
        """查询半年内活跃会员（消费 ∪ 备注 ∪ 任务变更）。"""

    async def run_member(self, member_id: int) -> None:
        """对单个会员执行调用链：
        App3→App8→App7
        如有助教关联：→App4→App5
        如有备注：→App6→App8（再次整合）
        验证 App5 输出包含分类字段（P2-1 修复）。
        """

    async def run(self) -> None:
        """主入口：加载断点 → 查询会员 → 分批执行 → 保存断点。"""

    def print_estimate(self, member_count: int) -> None:
        """输出预估信息：会员数、预估调用次数、预估 Token 消耗。"""
```


## 数据模型

### 已有表变更

#### biz.ai_run_logs — 新增字段（P15）

```sql
ALTER TABLE biz.ai_run_logs
    ADD COLUMN alert_status VARCHAR(20) DEFAULT NULL
    CHECK (alert_status IN ('pending', 'acknowledged', 'ignored'));

COMMENT ON COLUMN biz.ai_run_logs.alert_status IS
    'P15 新增 — 告警处理状态：NULL（非告警记录）/ pending（待处理）/ acknowledged（已确认）/ ignored（已忽略）。'
    '仅 status IN (''failed'',''timeout'',''circuit_open'') 的记录才设置此字段。';

-- 告警列表查询索引（筛选失败/超时/熔断 + alert_status）
CREATE INDEX idx_ai_run_logs_alert
    ON biz.ai_run_logs(alert_status, created_at DESC)
    WHERE status IN ('failed', 'timeout', 'circuit_open');
```

设计决策：在 `ai_run_logs` 上新增 `alert_status` 字段而非创建独立告警表。理由：
- 告警来源就是 `ai_run_logs` 中的失败/超时/熔断记录，1:1 关系
- 独立表需要维护外键和数据同步，增加复杂度
- 单字段方案查询简单（一次 WHERE 即可），无需 JOIN
- 绝大多数记录（成功调用）的 `alert_status` 为 NULL，不占索引空间（部分索引）

#### biz.ai_run_logs — 新增 BRIN 索引（P15）

```sql
-- Dashboard 聚合查询优化：BRIN 索引适合按时间顺序插入的表
CREATE INDEX idx_ai_run_logs_created_brin
    ON biz.ai_run_logs USING BRIN (created_at)
    WITH (pages_per_range = 32);

-- 可选：删除原有 B-tree 索引（如果 BRIN 索引足够覆盖查询）
-- DROP INDEX IF EXISTS biz.idx_ai_run_logs_created;
```

注意：保留原有 `idx_ai_run_logs_created` B-tree 索引，因为 Token 预算聚合（BudgetTracker）需要精确的等值/范围查询；BRIN 索引用于 Dashboard 的大范围时间聚合。

### 迁移脚本

文件：`db/zqyy_app/migrations/2026-03-23__p15_ai_monitoring.sql`

```sql
-- P15：AI 监控后台 — ai_run_logs 新增 alert_status 字段 + BRIN 索引
-- 依赖：P14 迁移（2026-03-22__p14_ai_module.sql）

BEGIN;

-- 1. 新增 alert_status 字段
ALTER TABLE biz.ai_run_logs
    ADD COLUMN IF NOT EXISTS alert_status VARCHAR(20) DEFAULT NULL;

ALTER TABLE biz.ai_run_logs
    ADD CONSTRAINT chk_ai_run_logs_alert_status
    CHECK (alert_status IS NULL OR alert_status IN ('pending', 'acknowledged', 'ignored'));

COMMENT ON COLUMN biz.ai_run_logs.alert_status IS
    'P15 — 告警处理状态：NULL/pending/acknowledged/ignored';

-- 2. 告警查询部分索引
CREATE INDEX IF NOT EXISTS idx_ai_run_logs_alert
    ON biz.ai_run_logs(alert_status, created_at DESC)
    WHERE status IN ('failed', 'timeout', 'circuit_open');

-- 3. BRIN 索引（Dashboard 聚合优化）
CREATE INDEX IF NOT EXISTS idx_ai_run_logs_created_brin
    ON biz.ai_run_logs USING BRIN (created_at)
    WITH (pages_per_range = 32);

-- 4. 回填已有失败记录的 alert_status
UPDATE biz.ai_run_logs
SET alert_status = 'pending'
WHERE status IN ('failed', 'timeout', 'circuit_open')
  AND alert_status IS NULL;

COMMIT;
```

### 回滚脚本

```sql
-- P15 回滚
BEGIN;
DROP INDEX IF EXISTS biz.idx_ai_run_logs_created_brin;
DROP INDEX IF EXISTS biz.idx_ai_run_logs_alert;
ALTER TABLE biz.ai_run_logs DROP CONSTRAINT IF EXISTS chk_ai_run_logs_alert_status;
ALTER TABLE biz.ai_run_logs DROP COLUMN IF EXISTS alert_status;
COMMIT;
```

### 数据保留策略汇总

| 数据类型 | 保留策略 | 清理方式 |
|---------|---------|---------|
| App1 对话记录（`ai_conversations` + `ai_messages`） | 永久保留 | 不清理 |
| App2~8 缓存（`ai_cache`） | 每个 App 保留最新 20,000 条 | 定时任务：按 `created_at` 排序，超出部分 DELETE |
| AI 运行记录（`ai_run_logs`） | 保留 90 天 | 定时任务：`DELETE WHERE created_at < now() - interval '90 days'` |
| 调度记录（`ai_trigger_jobs`） | 保留 90 天 | 同上 |

清理定时任务：每日凌晨 03:00 执行，通过 `scheduled_tasks` 表注册。


## 正确性属性

*属性（Property）是在系统所有合法执行路径上都应成立的特征或行为——本质上是对"系统应该做什么"的形式化陈述。属性是人类可读规格说明与机器可验证正确性保证之间的桥梁。*

> **与 P14 属性测试的关系**：P14 已实现 20 个属性测试（见 `.kiro/specs/P14-ai-dashscope-migration/design.md`）。P15 的 7 个 PRD 属性测试中，5 个是对 P14 已有测试的更新/增强（标注 `P14 更新`），2 个是 P15 全新属性。此外 P15 新增 7 个监控后台相关属性。

---

### PRD §3.3 指定的 7 个属性测试

### Property 1: 熔断器状态机 — 状态转换合法性（P14 Property 6 更新）

*对于任意* app_id 和任意成功/失败事件序列，熔断器状态转换应满足：连续 5 次失败→OPEN；OPEN 经过 recovery_timeout→HALF_OPEN；HALF_OPEN + 成功→CLOSED；HALF_OPEN + 失败→OPEN。任意操作序列不产生非法状态（即状态值始终为 closed/open/half_open 之一）。

**Validates: Requirements C3.1**

> P15 更新点：增强测试覆盖，确保 P14 迁移后熔断器在新 DashScope 调用链中行为一致。

### Property 2: Token 预算计算 — 日/月聚合不变量（P14 Property 8 更新）

*对于任意* 一组 `ai_run_logs` 记录（含 `tokens_used` 和 `created_at`），日聚合值应等于当日所有 `status='success'` 记录的 `tokens_used` 之和，月聚合值应等于当月之和。Dashboard API 和 Budget API 返回的聚合值应与直接 SQL 聚合结果一致。

**Validates: Requirements C3.2, A6.2**

> P15 更新点：验证 AdminAIService 的聚合查询与 BudgetTracker 的聚合结果一致。

### Property 3: 去重逻辑 — 唯一性不变量（P14 Property 12 更新）

*对于任意* 两个具有相同 `(event_type, member_id, site_id, date)` 的自动触发事件，第二个应被跳过（`skipped_duplicate`）；但当 `is_forced=true` 时（如 admin-web 手动重跑），即使存在重复也应正常执行。

**Validates: Requirements C3.3, A3.2**

> P15 更新点：验证 admin-web 手动重跑（is_forced=true）正确绕过去重检查。

### Property 4: 缓存过期 — expires_at 过期状态一致性（P15 新增）

*对于任意* `ai_cache` 记录，当 `expires_at < now()` 时，该记录的 `status` 应为 `expired`（由定时标记或查询时懒标记）。查询缓存时，过期记录不应被返回为有效缓存。

**Validates: Requirements C3.4**

### Property 5: App8 幂等 — member_retention_clue 唯一性（P14 Property 13 更新）

*对于任意* member_id 和日期，多次执行 App8 写入 `member_retention_clue` 后，该 member 该天的记录数应恒为 1（DELETE + INSERT 事务保证）。

**Validates: Requirements C3.5**

> P15 更新点：验证回填脚本中的 App8 写入同样满足幂等性。

### Property 6: session_id 重建 — round-trip 属性（P15 新增）

*对于任意* App1 对话（含本地 `ai_messages` 记录），当百炼 session_id 过期后重建会话时，重建后的 messages 数组应与本地 `ai_messages` 表中的记录一致（内容和顺序均匹配）。

**Validates: Requirements C3.6**

### Property 7: 限流计数 — 窗口内请求数上限（P14 Property 7 更新）

*对于任意* 用户 ID 或门店 ID，在滑动窗口内（App1: 60 秒/10 次，App2~8: 3600 秒/100 次）请求次数未超过阈值时应允许，超过阈值时应拒绝；窗口外的历史请求不影响当前判断。

**Validates: Requirements C3.7**

> P15 更新点：确保限流器在高并发回填场景下行为正确。

---

### P15 监控后台新增属性

### Property 8: Dashboard 聚合正确性

*对于任意* 一组 `ai_run_logs` 记录和任意 `site_id` 筛选条件，Dashboard API 返回的统计数据应满足：
- `today_calls` = 今日记录总数
- `today_success_rate` = 今日 success 记录数 / 今日记录总数
- `today_tokens` = 今日所有记录的 `tokens_used` 之和
- `today_avg_latency_ms` = 今日所有记录的 `latency_ms` 平均值
- `trend_7d` 中每天的 `calls` 和 `success_rate` 与按日分组聚合结果一致
- `app_distribution` 中各 App 的 `count` 之和等于总调用数，`percentage` 之和约等于 1.0

**Validates: Requirements A1.1, A1.2, A1.3, A1.6**

### Property 9: 告警筛选与状态转换

*对于任意* 一组 `ai_run_logs` 记录，告警列表应仅包含 `status IN ('failed', 'timeout', 'circuit_open')` 的记录。*对于任意* 告警记录，执行 ack 操作后 `alert_status` 应为 `acknowledged`，执行 ignore 操作后应为 `ignored`。`alert_status` 的值域始终为 `{NULL, 'pending', 'acknowledged', 'ignored'}`。

**Validates: Requirements A8.1, A8.2, A8.3, A8.4**

### Property 10: 缓存失效精确性

*对于任意* `ai_cache` 数据集和任意筛选条件组合（`site_id` 必填，`app_type` 和 `member_id` 可选），执行缓存失效后：匹配条件的记录 `status` 应为 `invalidated`，不匹配条件的记录 `status` 应保持不变。返回的 `affected_count` 应等于实际被更新的记录数。

**Validates: Requirements A5.1, A5.2, A5.3**

### Property 11: 批量执行预估公式

*对于任意* `app_types` 列表和 `member_ids` 列表，`estimated_calls` 应等于 `len(app_types) × len(member_ids)`，`estimated_tokens` 应等于 `estimated_calls × avg_tokens_per_call`（其中 `avg_tokens_per_call` 为系统配置常量，默认 2000）。

**Validates: Requirements A7.1, A7.2**

### Property 12: 分页与筛选正确性

*对于任意* `ai_run_logs` 或 `ai_trigger_jobs` 数据集、任意筛选条件组合和分页参数（page, page_size），返回的 `items` 应满足：
- 所有 item 均匹配所有指定的筛选条件
- `len(items) <= page_size`
- `total` 等于匹配筛选条件的总记录数
- 当指定 `site_id` 时，所有 item 的 `site_id` 均等于指定值

**Validates: Requirements A2.1, A2.2, A4.1, A4.2, A1.6**

### Property 13: 回填断点续跑 round-trip

*对于任意* 已完成的 member_id 集合，保存到 checkpoint 文件后重新加载，得到的集合应与原始集合完全一致（序列化/反序列化 round-trip）。

**Validates: Requirements D1.4**

### Property 14: 数据保留不变量

*对于任意* 数据库状态，执行清理任务后应满足：
- `ai_run_logs` 中不存在 `created_at < now() - 90 days` 的记录
- `ai_trigger_jobs` 中不存在 `created_at < now() - 90 days` 的记录
- 每个 App 类型（App2~8）的 `ai_cache` 记录数不超过 20,000
- `ai_conversations` 和 `ai_messages` 记录数不减少（永久保留）

**Validates: Requirements E1.1, E1.2, E1.3, E1.4**


## 错误处理

### Admin AI API 错误处理

| 场景 | HTTP 状态码 | 响应 |
|------|------------|------|
| JWT Token 缺失或无效 | 401 | `{"code": 401, "message": "未授权"}` |
| 调度任务不存在 | 404 | `{"code": 404, "message": "调度任务不存在"}` |
| 调用记录不存在 | 404 | `{"code": 404, "message": "调用记录不存在"}` |
| 告警记录不存在或非告警状态 | 404 | `{"code": 404, "message": "告警记录不存在"}` |
| batch_id 无效或已过期 | 400 | `{"code": 400, "message": "批量执行请求无效或已过期"}` |
| batch_id 已确认（重复确认） | 409 | `{"code": 409, "message": "批量执行已确认"}` |
| 缓存失效缺少 site_id | 422 | `{"code": 422, "message": "site_id 为必填参数"}` |
| 筛选参数格式错误（日期等） | 422 | `{"code": 422, "message": "参数格式错误: ..."}` |
| Dashboard 聚合查询超时 | 504 | `{"code": 504, "message": "查询超时，请缩小时间范围"}` |
| 数据库连接失败 | 500 | `{"code": 500, "message": "服务内部错误"}` |

### 回填脚本错误处理

| 场景 | 处理方式 |
|------|---------|
| 单个会员调用链失败 | 记录错误日志，跳过该会员，继续下一个；不写入 checkpoint |
| App8 写入 member_retention_clue 失败 | 事务回滚，记录错误到 ai_trigger_jobs |
| 数据库连接中断 | 保存当前 checkpoint，退出脚本，下次从断点继续 |
| App5 输出缺少分类字段 | 记录警告日志，不中断链路 |
| 预算超限 | 保存 checkpoint，输出提示信息，退出脚本 |

### 清理任务错误处理

| 场景 | 处理方式 |
|------|---------|
| 单步清理失败（如 ai_run_logs 删除失败） | 记录错误日志，继续执行后续步骤 |
| 清理过程中数据库锁等待超时 | 设置 `statement_timeout = 300000`（5 分钟），超时后跳过 |
| 清理任务整体失败 | 记录错误到 scheduled_tasks 的执行历史，下次定时触发重试 |

### 降级策略

- Dashboard 聚合查询超时：前端展示"数据加载中"，建议缩小时间范围
- 批量执行中某个会员失败：跳过该会员，继续执行其余会员，最终返回成功/失败统计
- 告警 ack/ignore 失败：前端提示"操作失败，请重试"


## 测试策略

### 属性测试（Property-Based Testing）

使用 `hypothesis` 库（项目已有依赖），每个属性测试最少运行 100 次迭代。

测试文件位于 `tests/` 目录（Monorepo 级属性测试），按组件分文件：

| 测试文件 | 覆盖属性 | P14/P15 关系 |
|---------|---------|-------------|
| `tests/test_circuit_breaker_props.py` | Property 1（熔断器状态机） | P14 Property 6 更新 |
| `tests/test_budget_tracker_props.py` | Property 2（Token 预算聚合） | P14 Property 8 更新 |
| `tests/test_dispatcher_props.py` | Property 3（去重唯一性） | P14 Property 12 更新 |
| `tests/test_cache_service_props.py` | Property 4（缓存过期一致性） | P15 新增 |
| `tests/test_app8_idempotent.py` | Property 5（App8 幂等） | P14 Property 13 更新 |
| `tests/test_session_props.py` | Property 6（session_id 重建 round-trip） | P15 新增 |
| `tests/test_rate_limiter_props.py` | Property 7（限流窗口） | P14 Property 7 更新 |
| `tests/test_admin_ai_dashboard_props.py` | Property 8（Dashboard 聚合） | P15 新增 |
| `tests/test_admin_ai_alert_props.py` | Property 9（告警筛选与状态） | P15 新增 |
| `tests/test_admin_ai_cache_props.py` | Property 10（缓存失效精确性） | P15 新增 |
| `tests/test_admin_ai_batch_props.py` | Property 11（批量预估公式） | P15 新增 |
| `tests/test_admin_ai_pagination_props.py` | Property 12（分页与筛选） | P15 新增 |
| `tests/test_backfill_props.py` | Property 13（断点续跑 round-trip） | P15 新增 |
| `tests/test_ai_cleanup_props.py` | Property 14（数据保留不变量） | P15 新增 |

每个属性测试必须包含注释标签：
```python
# Feature: ai-monitoring-testing, Property 1: 熔断器状态机 — 状态转换合法性
@given(st.lists(st.sampled_from(["success", "failure"]), min_size=1, max_size=50))
@settings(max_examples=100)
def test_circuit_breaker_state_machine(events):
    ...
```

### 单元测试

单元测试覆盖属性测试不适合的场景（具体示例、集成点、边界条件）：

| 测试文件 | 覆盖内容 |
|---------|---------|
| `apps/backend/tests/unit/test_admin_ai_router.py` | 13 个 API 端点的请求/响应格式、JWT 认证拒绝、参数校验 |
| `apps/backend/tests/unit/test_admin_ai_service.py` | AdminAIService 各方法的边界条件（空数据集、单条记录等） |
| `apps/backend/tests/unit/test_ai_cleanup_service.py` | 清理服务的边界条件（无数据、刚好 20000 条、跨日边界） |
| `apps/backend/tests/unit/test_backfill_script.py` | 回填脚本的 checkpoint 文件读写、空会员列表、批次划分 |

### 全链路测试（模块 C2）

文件：`apps/backend/tests/integration/test_ai_full_chain.py`

15 个场景的全链路测试，支持 Mock 模式和真实 API 模式双轨：

```python
# 通过环境变量切换模式
AI_TEST_MODE = os.environ.get("AI_TEST_MODE", "mock")  # mock / real

# Mock 模式：mock DashScopeClient.call_app() 返回预设 JSON
# Real 模式：调用真实百炼 API（需要 DASHSCOPE_API_KEY）
```

| 测试函数 | 场景 | 模式 |
|---------|------|------|
| `test_app1_10_entries` | App1 对话 10 种入口 | Mock + Real |
| `test_app2_pregenerate` | App2 定时预生成 8 维度 | Mock + Real |
| `test_consumption_chain` | 消费事件链 App3→App8→App7 | Mock + Real |
| `test_coach_consumption_chain` | 助教消费链 App3→App8→App7→App4→App5 | Mock + Real |
| `test_note_chain` | 备注事件链 App6→App8 | Mock + Real |
| `test_task_chain` | 任务分配链 App4→App5 | Mock + Real |
| `test_cache_hit` | 缓存命中 | Mock |
| `test_cache_invalidation` | 缓存失效后重新生成 | Mock |
| `test_circuit_breaker_flow` | 熔断触发→半开→恢复 | Mock |
| `test_budget_exceeded` | Token 预算超限降级 | Mock |
| `test_failure_logging` | 失败记录完整性 | Mock |
| `test_admin_visibility` | admin-web API 可见性 | Mock |
| `test_json_fallback` | JSON 兜底重试 3 次 | Mock |
| `test_app8_idempotent` | App8 幂等验证 | Mock |
| `test_content_quality` | 内容质量（必要字段+格式） | Mock + Real |

### 测试配置

```python
# conftest.py 中的 hypothesis 配置
from hypothesis import settings
settings.register_profile("ci", max_examples=200, deadline=10000)
settings.register_profile("dev", max_examples=100, deadline=5000)
```

### 测试依赖

- 属性测试：`hypothesis`（已有）
- Mock：`unittest.mock`（标准库）
- DashScope 调用：mock `Application.call()`，不依赖真实 API
- 数据库：使用 `test_zqyy_app` 测试库（遵循 `testing-env.md` 规范）
- 全链路测试：`AI_TEST_MODE` 环境变量控制 mock/real 模式


## 涉及文件汇总

### 模块 A：后端 API（新增）

| 文件路径 | 操作 | 说明 |
|---------|------|------|
| `apps/backend/app/routers/admin_ai.py` | 新建 | 13 个 admin AI API 端点 |
| `apps/backend/app/schemas/admin_ai.py` | 新建 | 请求/响应 Pydantic Schema |
| `apps/backend/app/services/ai/admin_service.py` | 新建 | AdminAIService 聚合服务 |
| `apps/backend/app/services/ai/cleanup_service.py` | 新建 | AICleanupService 数据清理 |
| `apps/backend/app/main.py` | 修改 | 注册 admin_ai.router |

### 模块 B：前端页面（新增）

| 文件路径 | 操作 | 说明 |
|---------|------|------|
| `apps/admin-web/src/api/adminAI.ts` | 新建 | Admin AI API 调用封装 |
| `apps/admin-web/src/pages/AIDashboard.tsx` | 新建 | AI 运行总览页面 |
| `apps/admin-web/src/pages/AITriggerJobs.tsx` | 新建 | 调度状态页面 |
| `apps/admin-web/src/pages/AIRunLogs.tsx` | 新建 | 调用明细页面 |
| `apps/admin-web/src/pages/AIOperations.tsx` | 新建 | 手动操作页面 |
| `apps/admin-web/src/App.tsx` | 修改 | 新增 4 个路由 + 侧边栏菜单项 |

### 模块 C：测试体系（归档 + 新增）

| 文件路径 | 操作 | 说明 |
|---------|------|------|
| `scripts/ops/_archived/ai_full_chain_test.py` | 移入 | 从 `scripts/ops/` 归档 |
| `scripts/ops/_archived/test_chat_e2e.py` | 移入 | 从 `scripts/ops/` 归档 |
| `scripts/ops/_archived/test_chat_ai_quality.py` | 移入 | 从 `scripts/ops/` 归档 |
| `scripts/ops/_archived/test_bailian_apps.py` | 移入 | 从 `scripts/ops/` 归档 |
| `scripts/ops/_archived/test_bailian_single.py` | 移入 | 从 `scripts/ops/` 归档 |
| `scripts/ops/_archived/test_bailian_full.py` | 移入 | 从 `scripts/ops/` 归档 |
| `scripts/ops/_archived/_run_ai_tests_remaining.py` | 移入 | 从 `scripts/ops/` 归档 |
| `apps/backend/tests/_archived/test_ai_bailian.py` | 移入 | 从 `apps/backend/tests/` 归档 |
| `docs/reports/_archived/2026-03-21__ai_full_chain_test.md` | 移入 | 从 `docs/reports/` 归档 |
| `apps/backend/tests/integration/test_ai_full_chain.py` | 新建 | 15 场景全链路测试 |
| `tests/test_circuit_breaker_props.py` | 修改 | Property 1 更新 |
| `tests/test_budget_tracker_props.py` | 修改 | Property 2 更新 |
| `tests/test_dispatcher_props.py` | 修改 | Property 3 更新 |
| `tests/test_cache_service_props.py` | 修改 | Property 4 新增 |
| `tests/test_app8_idempotent.py` | 修改 | Property 5 更新 |
| `tests/test_session_props.py` | 修改 | Property 6 新增 |
| `tests/test_rate_limiter_props.py` | 修改 | Property 7 更新 |
| `tests/test_admin_ai_dashboard_props.py` | 新建 | Property 8 |
| `tests/test_admin_ai_alert_props.py` | 新建 | Property 9 |
| `tests/test_admin_ai_cache_props.py` | 新建 | Property 10 |
| `tests/test_admin_ai_batch_props.py` | 新建 | Property 11 |
| `tests/test_admin_ai_pagination_props.py` | 新建 | Property 12 |
| `tests/test_backfill_props.py` | 新建 | Property 13 |
| `tests/test_ai_cleanup_props.py` | 新建 | Property 14 |

### 模块 D：回填脚本（新增）

| 文件路径 | 操作 | 说明 |
|---------|------|------|
| `scripts/ops/ai_backfill.py` | 新建 | 历史数据回填脚本 |

### 模块 E：数据保留（新增）

| 文件路径 | 操作 | 说明 |
|---------|------|------|
| `apps/backend/app/services/ai/cleanup_service.py` | 新建 | 同模块 A（清理服务） |

### 模块 F：查询优化与收尾

| 文件路径 | 操作 | 说明 |
|---------|------|------|
| `db/zqyy_app/migrations/2026-03-23__p15_ai_monitoring.sql` | 新建 | DDL 迁移（alert_status + BRIN 索引） |
| `docs/database/ddl/zqyy_app__biz.sql` | 修改 | 合并 P15 DDL 到基线 |
| `docs/database/BD_Manual_ai_tables.md` | 修改 | 补充 alert_status 字段 + admin API 查询模式 |
| `apps/admin-web/README.md` | 修改 | 新增 AI 监控页面说明 |
| `apps/backend/README.md` | 修改 | 新增 admin AI API 说明 |
| `docs/DOCUMENTATION-MAP.md` | 修改 | 新增 P15 相关条目 |

### 单元测试（新增）

| 文件路径 | 操作 | 说明 |
|---------|------|------|
| `apps/backend/tests/unit/test_admin_ai_router.py` | 新建 | API 端点单元测试 |
| `apps/backend/tests/unit/test_admin_ai_service.py` | 新建 | AdminAIService 单元测试 |
| `apps/backend/tests/unit/test_ai_cleanup_service.py` | 新建 | 清理服务单元测试 |
| `apps/backend/tests/unit/test_backfill_script.py` | 新建 | 回填脚本单元测试 |