feat(bot-tier): M1 Bot侧高危建议创建通道+目标值直写收口 (P0-②分层授权)

- 新增 POST /api/cma/bot/ai/suggestions (bot_suggestions.py, X-BOT-KEY鉴权, L2)
  高危建议(kpi_target/budget_adjust/action_plan) status恒unapplied 等人审批
  source白名单(bot/rule)防冒充manual; entity多租户归属校验; 幂等防重
- risk_levels.py: API_RISK_MAP加新端点 + 文件头补写操作语义分级low/high说明
- 收口: bot/kpis/create-with-links 拒绝直写非零target_value(显式置0, 考核基准=人拍板)
- test_risk_levels: L2计数5→6; 新增test_bot_suggestion_channel 12条(鉴权/语义/多租户/幂等/审批闭环/审计)

backend/app/api/bot_bridge.py

@@ -728,11 +728,26 @@ def bot_kpi_value_with_check(data: dict, db: Session = Depends(get_db), bot: dic
 @risk_level("L3")
 def bot_kpi_create_with_links(data: dict, db: Session = Depends(get_db), bot: dict = Depends(verify_bot_key)):
     """聚合B: 创建KPI + 关联战略地图 + 批量因果链（Agent建KPI标准动作）
-    body: {kpi_code, kpi_name, dimension, entity_id?, target_value?, unit?, link_map_id?, link_causality?}"""
+    body: {kpi_code, kpi_name, dimension, entity_id?, unit?, link_map_id?, link_causality?}
+
+    分层授权收口（2026-09-03 P0-②）: Bot 创建KPI不得携带自定目标值——
+    考核基准类（目标值）= 人拍板；target_value 一律显式置 0（满足治理必填，
+    后续由人工设定或走 POST /api/cma/bot/ai/suggestions 建议通道待审批）。
+    """
     entity_id = int(data.get("entity_id") or 1)
     from app.api.kpis import _validate_kpi_data, apply_calc_type_inference
     kpi_data = {k: v for k, v in data.items() if k not in ("entity_id", "link_map_id", "link_causality")}
 
+    # 收口: 非零 target_value 直接拒绝（不静默丢弃，让调用方明确知晓规则）
+    tv_raw = data.get("target_value")
+    if tv_raw not in (None, 0, "0", 0.0):
+        raise HTTPException(
+            400,
+            "Bot创建KPI不允许直写目标值（考核基准类=人拍板，2026-09-03分层授权）。"
+            "请省略 target_value（自动置0，人工后续设定）或改走 POST /api/cma/bot/ai/suggestions 建议通道待审批。",
+        )
+    kpi_data["target_value"] = 0  # 显式置0（覆盖 0/"0"/None 三种传入）
+
     # ① 创建KPI（编码唯一 + 治理校验 + 强制企业）
     code = kpi_data.get("kpi_code", "")
     if not code:

backend/app/api/bot_suggestions.py

@@ -0,0 +1,175 @@
+"""Bot侧AI高危建议创建通道 — 写操作分层授权落地 (P0-② 2026-09-03)
+
+背景（CMA交互架构规范 v1 5.1 分层授权模型，任总拍板）:
+    考核基准类（KPI目标值/预算调整）→ 🔒 Bot只能"建议"，人审批后生效
+    Bot不得直写 KPI target_value / budget 审批类字段；
+    Bot产生的高危变更一律走本通道建 ai_suggestions（status=unapplied）
+    → UI AI建议中心(SuggestionCenter.vue)待批列表 → 人 apply 后生效+留痕。
+
+通道关系（两套分级勿混，见 risk_levels.py 顶部注释）:
+    L1-L4（技术风险分级）: 本端点挂 L2（业务写单条）
+    low/high（业务写语义分级）: 本端点是 high 类的"建议载体"——
+        端点本身不直写考核基准，只建建议行，属 high 流程的入口。
+
+鉴权/审计:
+    X-BOT-KEY（verify_bot_key, bot_bridge 共享密钥）+ @risk_level("L2")
+    路径 /api/cma/bot/* → bot_audit_middleware 自动留痕（bot_name/risk/status）
+    entity_id 多租户: X-Entity-Id header → body.entity_id → 默认1，校验 Entity.active
+"""
+from typing import Optional
+
+from fastapi import APIRouter, Depends, HTTPException, Request
+from sqlalchemy.orm import Session
+
+from app.database import get_db
+from app.models import AISuggestion, KPIDefinition, Entity
+from app.api.bot_bridge import verify_bot_key
+from app.api.ai_suggestions import _sug_dict
+from app.risk_levels import risk_level
+
+router = APIRouter(prefix="/api/cma/bot/ai/suggestions", tags=["AI建议Bot通道"])
+
+# 高危建议类型（Bot仅允许创建这三类建议，禁止其他）
+_SUPPORTED_TYPES = ("kpi_target", "budget_adjust", "action_plan")
+# source 白名单：Bot 不得冒充 manual/dashboard/kpi/budget（影响UI语义与推送判断）
+_ALLOWED_SOURCES = ("rule", "bot")
+_TARGET_TYPE_MAP = {
+    "kpi_target": "kpi",
+    "budget_adjust": "budget",
+    "action_plan": "kpi",
+}
+
+
+def _resolve_bot_entity(request: Request, data: dict, db: Session) -> int:
+    """账套解析（Bot无token通道）: body.entity_id → X-Entity-Id header → 默认1，校验active。
+
+    与 deps.get_entity_id 的 Bot 白名单语义一致（query/header/body 任一来源），
+    本端点不接收登录token，故不在此处理 Bearer。
+    """
+    raw = None
+    if data.get("entity_id") not in (None, ""):
+        raw = data["entity_id"]
+    else:
+        xh = request.headers.get("X-Entity-Id")
+        if xh and str(xh).isdigit():
+            raw = int(xh)
+    try:
+        eid = int(raw) if raw is not None else 1
+    except (TypeError, ValueError):
+        raise HTTPException(422, f"entity_id 无效: {raw!r}")
+    ent = db.query(Entity).filter(Entity.id == eid).first()
+    if not ent or ent.status != "active":
+        raise HTTPException(403, f"企业 entity_id={eid} 不存在或未激活")
+    return eid
+
+
+@router.post("")
+@risk_level("L2")
+def bot_create_suggestion(
+    request: Request,
+    data: dict,
+    bot: dict = Depends(verify_bot_key),
+    db: Session = Depends(get_db),
+):
+    """Bot创建高危建议（kpi_target/budget_adjust/action_plan），status 固定 unapplied 等人审批。
+
+    入参（字段与人工创建端点 /api/cma/ai/suggestions 一致）:
+      suggestion_type (str, 必填): kpi_target / budget_adjust / action_plan
+      title          (str, 必填): 建议标题（幂等键之一）
+      entity_id      (int, 可选, 默认1): 账套；也支持 X-Entity-Id header
+      target_id      (int, 可选): 目标ID；kpi_target/budget_adjust 必填(kpi_id 亦可)
+      suggestion_data(dict, 可选): 应用参数
+          kpi_target:   {kpi_id, target_value}
+          budget_adjust:{kpi_id, period:YYYY-MM, budget_value}
+          action_plan:  {kpi_id, title?, assignee?, priority?, due_date?}
+      content        (str, 可选): 建议理由/依据（财务Bot三情景测算content携带）
+      source         (str, 默认"bot"): 仅允许 bot/rule，禁止冒充 manual/dashboard
+      target_type    (str, 可选): 缺省按 suggestion_type 映射 kpi/budget
+
+    约束:
+      - status 恒为 unapplied（Bot 不能直接生效）
+      - kpi_target/budget_adjust 必须带有效 kpi_id/target_id 且属于当前账套
+      - 幂等: 同 entity+suggestion_type+target_id+title 且 unapplied 已存在 → 返回既有建议
+    """
+    suggestion_type = data.get("suggestion_type")
+    title = str(data.get("title") or "").strip()
+    if suggestion_type not in _SUPPORTED_TYPES:
+        raise HTTPException(
+            400,
+            f"Bot建议仅支持 suggestion_type: {'/'.join(_SUPPORTED_TYPES)}（高危变更=建议通道，不提供直写）",
+        )
+    if not title:
+        raise HTTPException(400, "缺少 title")
+
+    source = data.get("source", "bot")
+    if source not in _ALLOWED_SOURCES:
+        raise HTTPException(400, f"source 仅允许 {'/'.join(_ALLOWED_SOURCES)}（禁止冒充 manual/dashboard/kpi/budget）")
+
+    entity_id = _resolve_bot_entity(request, data, db)
+    sd = data.get("suggestion_data") or {}
+    target_type = data.get("target_type") or _TARGET_TYPE_MAP.get(suggestion_type)
+
+    # 目标归属校验（多租户铁律: 只能建议当前账套的KPI）
+    kpi_id = None
+    raw_kpi = data.get("kpi_id") or data.get("target_id") or sd.get("kpi_id")
+    if raw_kpi is not None:
+        try:
+            kpi_id = int(raw_kpi)
+        except (TypeError, ValueError):
+            kpi_id = None
+    if suggestion_type in ("kpi_target", "budget_adjust"):
+        if kpi_id is None:
+            raise HTTPException(400, f"suggestion_type={suggestion_type} 必须带 kpi_id/target_id")
+        kpi = db.query(KPIDefinition).filter(
+            KPIDefinition.id == kpi_id,
+            KPIDefinition.entity_id == entity_id,
+        ).first()
+        if not kpi:
+            raise HTTPException(404, f"KPI {kpi_id} 不存在于当前账套 entity_id={entity_id}")
+        if suggestion_type == "kpi_target" and sd.get("target_value") is None:
+            raise HTTPException(400, "kpi_target 建议的 suggestion_data 必须带 target_value")
+        if suggestion_type == "budget_adjust" and (
+            not sd.get("period") or sd.get("budget_value") is None
+        ):
+            raise HTTPException(400, "budget_adjust 建议的 suggestion_data 必须带 period(YYYY-MM) + budget_value")
+
+    # 幂等: 同 entity+type+target_id+title 的 unapplied 建议存在则不重复创建
+    target_id = kpi_id if kpi_id is not None else data.get("target_id")
+    dup = db.query(AISuggestion).filter(
+        AISuggestion.entity_id == entity_id,
+        AISuggestion.suggestion_type == suggestion_type,
+        AISuggestion.target_id == target_id,
+        AISuggestion.title == title,
+        AISuggestion.status == "unapplied",
+    ).first()
+    if dup:
+        return {
+            "success": True,
+            "message": "建议已存在（幂等，未重复创建）",
+            "data": _sug_dict(dup),
+            "duplicated": True,
+        }
+
+    category = "alert" if target_type == "alert" else data.get("category", "decision")
+    sug = AISuggestion(
+        entity_id=entity_id,
+        user_id=None,  # Bot 创建无人工账号；bot 名与调用留痕在 bot_audit.log（X-BOT-KEY）
+        source=source,
+        suggestion_type=suggestion_type,
+        target_type=target_type,
+        category=category,
+        target_id=target_id,
+        title=title,
+        content=data.get("content"),
+        suggestion_data=sd,
+        status="unapplied",  # 铁律: Bot 只能建议，生效必须人 apply
+    )
+    db.add(sug)
+    db.commit()
+    db.refresh(sug)
+    return {
+        "success": True,
+        "message": "高危建议已创建，等待人工审批（AI建议中心 → 应用）",
+        "data": _sug_dict(sug),
+        "duplicated": False,
+    }

backend/app/main.py

@@ -5,7 +5,7 @@ from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import JSONResponse
 from dotenv import load_dotenv
 from app.database import init_db
-from app.api import auth, kpis, kpi_governance, templates, maps, dashboard, data, alerts, ai_analysis, ai_suggestions, alert_rules, users, thresholds, notifications, permissions, action_plans, alignment, org, objectives, versions, budget, cost, predict, reports, security, knowledge, bot_bridge, bot_bridge_v2, lead, tenant, customer_dashboard, deviation_push, budget_generate, knowledge_articles, kpi_causality, data_quality, bi_reports, entities, bsc_layers, okr, okr_templates, subjects, driver_budget, bot_kpis, ontology, bot_iron_law, analysis_results, expenses, cash, tax_compliance, verify, growth_quality, products, data_classification, value_sources, zero_based, derivation_rules, cash_classify, cashflow_structure
+from app.api import auth, kpis, kpi_governance, templates, maps, dashboard, data, alerts, ai_analysis, ai_suggestions, alert_rules, users, thresholds, notifications, permissions, action_plans, alignment, org, objectives, versions, budget, cost, predict, reports, security, knowledge, bot_bridge, bot_bridge_v2, bot_suggestions, lead, tenant, customer_dashboard, deviation_push, budget_generate, knowledge_articles, kpi_causality, data_quality, bi_reports, entities, bsc_layers, okr, okr_templates, subjects, driver_budget, bot_kpis, ontology, bot_iron_law, analysis_results, expenses, cash, tax_compliance, verify, growth_quality, products, data_classification, value_sources, zero_based, derivation_rules, cash_classify, cashflow_structure
 from app.utils.cache import clear_all as clear_cache, delete as delete_cache
 from scripts.erp_sync import run_sync as run_erp_sync
 from app.auth_middleware import require_auth
@@ -44,6 +44,7 @@ app.include_router(data.router)
 app.include_router(alerts.router)
 app.include_router(ai_analysis.router)
 app.include_router(ai_suggestions.router)
+app.include_router(bot_suggestions.router)
 app.include_router(alert_rules.router)
 app.include_router(users.router)
 app.include_router(thresholds.router)

backend/app/risk_levels.py

@@ -9,6 +9,34 @@ CMA Bot API 风险分级（L1-L4）与路由标注装饰器
     L4 危险           — 不向Bot API开放（DROP/TRUNCATE/批量DELETE/生产结构修改）
 
 安全底线: Bot API面不存在L4端点（0项开放 = 天然隔离）。
+
+写操作语义分级（业务语义 low/high，2026-09-03 分层授权落地 P0-②）
+============================================================
+依据: CMA交互架构规范 v1 第5.1节 分层授权模型（任总拍板）——
+  考核基准类（KPI目标值/预算调整）→ Bot只能"建议"，人拍板生效；
+  监测参数类（红黄绿阈值）→ Bot建议+人批量确认（P1待办，未开放Bot直写）；
+  预算 = 三阶段分权（编制人定框架/执行Bot自动/调整人审批）。
+
+low/high 与 L1-L4 是两套正交分级，勿混:
+  L1-L4 = 技术风险分级（操作面/写前校验/批量约束，本文件API_RISK_MAP）
+  low/high = 业务写语义分级（值是不是"考核基准/资源承诺"）
+  关系: low 类 = 现有 L2/L3 直写端点行为不变（规则型数据写）；
+        high 类 = Bot API面不提供直写端点，只提供建议通道端点（L2）。
+
+端点归类清单（low=Bot直写+留痕 / high=建议通道人审批）:
+  [low · 直写 KPI实际值/流水/验证记录]（写库即生效，靠 audit_log/source_batch 留痕）
+    POST /api/cma/bot/kpi-value-with-check           — 写KPI实际值+联动预警 (L2)
+    POST /api/cma/bot-bridge/kpi-result              — 写KPI实际值 (source_batch) (L2)
+    POST /api/cma/bot-kpis/{kpi_id}/value            — 写KPI当前值 (L2)
+    POST /api/cma/bot/import                         — Excel批量导入KPI实际值 (L3)
+    POST /api/cma/bot-bridge/mpm-result              — MPM分析结果→KPI实际值 (L3)
+    POST /api/cma/bot-bridge/verify/{action_plan_id} — 验证ActionPlan执行 (L2, 轻量verify_log)
+    POST /api/cma/bot/okr/create                     — 创建OKR Objective (L2, 管理类业务写，保留评审: 2026-09-03 P1待办)
+    POST /api/cma/bot/kpis/create-with-links         — 创建新KPI定义+因果链 (L3; 2026-09-03收口: 拒绝直写非零 target_value)
+  [high · 建议通道（端点只建 ai_suggestions, status=unapplied, 人 apply 后生效+apply_detail留痕）]
+    POST /api/cma/bot/ai/suggestions                 — Bot创建高危建议 (L2, 本任务新增)
+    说明: KPI目标值修改/预算调整 无任何Bot直写端点（grep实证）——
+          kpi_target/budget_adjust 的生效只经 /api/cma/ai/suggestions/{id}/apply（人token）。
 """
 import functools
 
@@ -54,12 +82,13 @@ API_RISK_MAP = {
     "GET /api/cma/bot-kpis": "L1",
     # 本任务新增的只读端点
     "GET /api/cma/bot/risk-levels": "L1",
-    # ── L2 业务写（4项方案清单 + okr/create单条业务写） ──
+    # ── L2 业务写（4项方案清单 + okr/create单条业务写 + bot建议通道） ──
     "POST /api/cma/bot/kpi-value-with-check": "L2",
     "POST /api/cma/bot-bridge/kpi-result": "L2",
     "POST /api/cma/bot-kpis/{kpi_id}/value": "L2",
     "POST /api/cma/bot-bridge/verify/{action_plan_id}": "L2",
     "POST /api/cma/bot/okr/create": "L2",  # 单条OKR创建（方案清单未列出，按单条业务写归类）
+    "POST /api/cma/bot/ai/suggestions": "L2",  # Bot高危建议创建通道（写操作语义 high，见文件头注释；本身只建建议不直写考核基准）
     # ── L3 批量写/创建（3项） ──
     "POST /api/cma/bot/import": "L3",
     "POST /api/cma/bot/kpis/create-with-links": "L3",

backend/tests/test_bot_suggestion_channel.py

@@ -0,0 +1,196 @@
+"""Bot侧高危建议创建通道 + 分层授权 测试 (P0-② 2026-09-03)
+
+覆盖:
+  1. 鉴权: 无 X-BOT-KEY → 401; X-BOT-KEY → 200
+  2. 高危语义: status 恒 unapplied; source 白名单(bot/rule); 类型白名单
+  3. 多租户: 建议只能指向当前账套KPI（跨账套 target → 404）; entity 不存在/未激活 → 403
+  4. 幂等: 同 entity+type+target_id+title 未应用已存在 → 不重复创建
+  5. 审批闭环: 人 apply → KPI目标变更 + apply_detail before/after + applied_by/applied_at + 重复apply拒绝
+  6. audit 留痕: bot 创建建议写入 audit（monkeypatch捕获 /api/cma/bot/* 中间件）
+  7. create-with-links 收口: Bot 直写非零 target_value → 400
+"""
+import json
+
+from tests.conftest import create_test_user, get_token_for_user, auth_header, create_test_kpi
+from app.models import AISuggestion, KPIDefinition, Entity
+
+BOT_KEY = {"X-BOT-KEY": "cma-bot-finance-2026"}
+
+BOT_PATH = "/api/cma/bot/ai/suggestions"
+
+
+def _bot_create(client, kpi_id, **kw):
+    body = {
+        "suggestion_type": "kpi_target",
+        "target_id": kpi_id,
+        "title": "Bot建议：上调目标（E2E-验证）",
+        "content": "规则引擎/财务Bot分析建议",
+        "source": "bot",
+        "suggestion_data": {"kpi_id": kpi_id, "target_value": 150.0},
+    }
+    body.update(kw)
+    return client.post(BOT_PATH, json=body, headers=BOT_KEY)
+
+
+class TestBotAuth:
+    def test_requires_bot_key(self, client, db):
+        kpi = create_test_kpi(db)
+        r = client.post(BOT_PATH, json={"suggestion_type": "kpi_target", "title": "x",
+                                        "target_id": kpi.id})
+        assert r.status_code == 401
+
+    def test_create_with_bot_key(self, client, db):
+        kpi = create_test_kpi(db)
+        r = _bot_create(client, kpi.id)
+        assert r.status_code == 200, r.text
+        data = r.json()["data"]
+        assert data["status"] == "unapplied"
+        assert data["source"] == "bot"
+        assert data["suggestion_type"] == "kpi_target"
+        assert data["user_id"] is None
+        # 落库
+        sug = db.query(AISuggestion).filter(AISuggestion.id == data["id"]).first()
+        assert sug is not None and sug.status == "unapplied"
+
+    def test_human_token_rejected_on_bot_path(self, client, db):
+        """人token不能走Bot通道（Bot端点只认X-BOT-KEY）"""
+        create_test_user(db)
+        token = get_token_for_user(client)
+        kpi = create_test_kpi(db)
+        r = client.post(BOT_PATH, json={"suggestion_type": "kpi_target", "title": "x",
+                                        "target_id": kpi.id, "target_value": 150},
+                        headers=auth_header(token))
+        assert r.status_code == 401
+
+
+class TestBotSemantics:
+    def test_type_whitelist(self, client, db):
+        kpi = create_test_kpi(db)
+        r = _bot_create(client, kpi.id, suggestion_type="delete_all")
+        assert r.status_code == 400
+        assert "仅支持" in r.text
+
+    def test_source_whitelist_no_spoof(self, client, db):
+        kpi = create_test_kpi(db)
+        r = _bot_create(client, kpi.id, source="manual")
+        assert r.status_code == 400
+        assert "source" in r.text
+
+    def test_kpi_target_requires_kpi_and_value(self, client, db):
+        # 无 kpi/target
+        r = _bot_create(client, None, suggestion_data={})
+        assert r.status_code == 400
+        # kpi_target 无 target_value
+        kpi = create_test_kpi(db)
+        r2 = _bot_create(client, kpi.id, suggestion_data={"kpi_id": kpi.id})
+        assert r2.status_code == 400
+
+    def test_cross_entity_target_rejected(self, client, db):
+        """多租户铁律: 建议只能指向当前账套KPI"""
+        db.add(Entity(id=2, name="另一企业", short_name="B", status="active"))
+        db.commit()
+        kpi_e1 = create_test_kpi(db, entity_id=1)
+        # 以 entity_id=2 身份建议 entity1 的KPI → 404
+        r = _bot_create(client, kpi_e1.id, entity_id=2)
+        assert r.status_code == 404
+        assert "不存在于当前账套" in r.text
+        # entity 不存在/未激活 → 403
+        r2 = _bot_create(client, kpi_e1.id, entity_id=99)
+        assert r2.status_code == 403
+
+    def test_idempotent(self, client, db):
+        kpi = create_test_kpi(db)
+        r1 = _bot_create(client, kpi.id)
+        r2 = _bot_create(client, kpi.id)
+        assert r1.status_code == 200
+        assert r2.status_code == 200
+        assert r2.json()["duplicated"] is True
+        assert r2.json()["data"]["id"] == r1.json()["data"]["id"]
+        assert db.query(AISuggestion).count() == 1
+
+
+class TestApproveClosedLoop:
+    def test_bot_suggestion_apply_chain(self, client, db):
+        """bot创建kpi_target建议 → unapplied → 人apply → 目标值生效 + 留痕完整"""
+        create_test_user(db)
+        token = get_token_for_user(client)
+        kpi = create_test_kpi(db, target_value=100.0)
+
+        r = _bot_create(client, kpi.id, title="Bot建议：上调目标（E2E-验证）")
+        sug_id = r.json()["data"]["id"]
+        assert r.json()["data"]["status"] == "unapplied"
+
+        # 待批列表可见（status=unapplied）
+        lst = client.get("/api/cma/ai/suggestions?status=unapplied", headers=auth_header(token)).json()
+        assert any(s["id"] == sug_id for s in lst["data"])
+
+        # 人 apply
+        app = client.post(
+            f"/api/cma/ai/suggestions/{sug_id}/apply",
+            json={"action": "kpi_target", "target_value": 150.0},
+            headers=auth_header(token),
+        )
+        assert app.status_code == 200, app.text
+        ad = app.json()["data"]
+        assert ad["status"] == "applied"
+        assert ad["applied_by"] == "测试管理员"
+        assert ad["applied_at"]
+        assert ad["apply_detail"][0]["before"] == 100.0
+        assert ad["apply_detail"][0]["after"] == 150.0
+        assert ad["apply_detail"][0]["action"] == "update_target_value"
+
+        db.refresh(kpi)
+        assert kpi.target_value == 150.0
+
+        # 重复apply拒绝
+        app2 = client.post(
+            f"/api/cma/ai/suggestions/{sug_id}/apply",
+            json={"action": "kpi_target", "target_value": 200.0},
+            headers=auth_header(token),
+        )
+        assert app2.status_code == 400
+
+
+class TestAuditCoverage:
+    def test_bot_suggestion_audited(self, client, db, monkeypatch):
+        """/api/cma/bot/ai/suggestions 被 bot_audit_middleware 留痕（risk L2）"""
+        import app.api.audit_log as audit_log_module
+        records = []
+        monkeypatch.setattr(audit_log_module, "write_audit_line",
+                            lambda rec: records.append(rec))
+        kpi = create_test_kpi(db)
+        r = _bot_create(client, kpi.id)
+        assert r.status_code == 200
+        assert records, "应有审计记录"
+        rec = records[-1]
+        assert rec["endpoint"] == BOT_PATH
+        assert rec["method"] == "POST"
+        assert rec["risk_level"] == "L2"
+        assert rec["bot_name"] == "财务BOT"
+        assert rec["status"] == 200
+
+
+class TestCreateWithLinksClosure:
+    def test_nonzero_target_value_rejected(self, client, db):
+        """收口: create-with-links 直写非零 target_value → 400"""
+        r = client.post("/api/cma/bot/kpis/create-with-links",
+                        json={"kpi_code": "E2E_TGT", "kpi_name": "收口验证KPI",
+                              "dimension": "finance", "target_value": 500000,
+                              "entity_id": 1},
+                        headers=BOT_KEY)
+        assert r.status_code == 400
+        assert "不允许直写目标值" in r.text
+
+    def test_without_target_value_ok(self, client, db):
+        """省略/置0 target_value 仍可创建KPI（收口自动置0，不阻断KPI创建）"""
+        r = client.post("/api/cma/bot/kpis/create-with-links",
+                        json={"kpi_code": "F_E2E_TGT2", "kpi_name": "收口放行KPI",
+                              "dimension": "finance", "unit": "万元", "formula": "营业收入",
+                              "data_source": "Bot创建验证", "data_owner": "财务部",
+                              "frequency": "monthly", "kpi_level": "operational",
+                              "entity_id": 1},
+                        headers=BOT_KEY)
+        assert r.status_code == 200, r.text
+        kpi = db.query(KPIDefinition).filter(KPIDefinition.kpi_code == "F_E2E_TGT2").first()
+        assert kpi is not None
+        assert kpi.target_value == 0

backend/tests/test_risk_levels.py

@@ -49,11 +49,11 @@ class TestRiskMapCoverage:
         assert not unlabeled, f"以下路由函数缺少 @risk_level 装饰器: {unlabeled}"
 
     def test_risk_level_counts(self):
-        """分级统计与方案一致：L1=21(20项清单+risk-levels端点)、L2=5(4项清单+okr/create)、L3=3、L4=0"""
+        """分级统计与方案一致：L1=21(20项清单+risk-levels端点)、L2=6(4项清单+okr/create+bot建议通道)、L3=3、L4=0"""
         from collections import Counter
         counts = Counter(API_RISK_MAP.values())
         assert counts["L1"] == 21, counts
-        assert counts["L2"] == 5, counts
+        assert counts["L2"] == 6, counts
         assert counts["L3"] == 3, counts
         assert counts["L4"] == 0, "Bot API面不得存在L4端点（安全底线）"
 
@@ -94,7 +94,7 @@ class TestRiskLevelsEndpoint:
 
         # 分级统计
         assert data["summary"]["L1"] == 21
-        assert data["summary"]["L2"] == 5
+        assert data["summary"]["L2"] == 6
         assert data["summary"]["L3"] == 3
         assert data["summary"]["L4"] == 0