fix(security): ai_analysis多租户隔离缺口 + dashboard预警计数 + bot_kpis归属 + cost账套过滤 + kpis create白名单 (OpenCode审查20260907)

- ai_analysis: generate_rule_suggestions pending预警按entity过滤; kpi_analysis归属校验404;
  dashboard-analysis-stream/ask/review-plans 补 get_entity_id + KPI/ActionPlan/KPIAlert 全按账套过滤;
  stream cache key 含 entity_id (原跨企业串缓存)
- dashboard: summary alert_count 补 KPIAlert.entity_id 过滤 (kpi_total 已有)
- bot_kpis: update_bot_kpi_value 归属校验 404 (原任意登录可改他企 bot KPI)
- cost: ActualCost/StandardCost 确认多企业共用 → cost_overview/variance/breakdown/dashboard
  补 entity 过滤(engine 函数加 entity_id 透传); dashboard total_cost 既有 KeyError 修复
- kpis: create_kpi 字段白名单 (仿 UPDATE_KPI_WHITELIST+kpi_code), 拦截 status/important_flag 注入
- tests: 新增 test_multitenant_ai_20260907.py 11条越权用例 (跨entity 404/空返回断言)
- 全量 pytest: 745 passed / 10 failed(与基线 HEAD 一致, 存量问题非本次引入) / 1 xfailed

backend/app/api/ai_analysis.py

@@ -156,10 +156,16 @@ def generate_rule_suggestions(db: Session, entity_id: int,
                 "kpi_id": b.kpi_id, "period": period, "budget_value": round(actual, 2),
             })
 
-    # 4. pending预警 → 建行动方案
-    alerts = db.query(KPIAlert).filter(KPIAlert.status == "pending").all()
+    # 4. pending预警 → 建行动方案（多租户隔离 2026-09-07: 只取本账套预警）
+    alerts = db.query(KPIAlert).filter(
+        KPIAlert.entity_id == entity_id,
+        KPIAlert.status == "pending",
+    ).all()
     for a in alerts:
-        k = db.query(KPIDefinition).filter(KPIDefinition.id == a.kpi_id).first()
+        k = db.query(KPIDefinition).filter(
+            KPIDefinition.id == a.kpi_id,
+            KPIDefinition.entity_id == entity_id,
+        ).first()
         kpi_name = k.kpi_name if k else f"KPI#{a.kpi_id}"
         title = f"处理预警：{kpi_name} {a.alert_message[:30]}"
         content = f"存在待处理预警（{a.alert_level}级）：{a.alert_message}。建议建立行动方案跟进。"
@@ -291,8 +297,11 @@ async def dashboard_analysis(role: str = Query("ceo"), db: Session = Depends(get
             "unit": k.unit,
         })
     
-    # 获取预警
-    alerts = db.query(KPIAlert).filter(KPIAlert.status == "pending").count()
+    # 获取预警（多租户隔离 2026-09-07: 只统计本账套）
+    alerts = db.query(KPIAlert).filter(
+        KPIAlert.entity_id == entity_id,
+        KPIAlert.status == "pending",
+    ).count()
     
     # 构建分析prompt
     kpi_text = "\n".join([f"- {k['name']}({k['code']}): 目标={k['target']}, 实际={k['actual']}({k['period']}), 维度={k['dimension']}" for k in kpi_summary if k['actual'] is not None])
@@ -339,8 +348,12 @@ async def kpi_analysis(kpi_id: int, db: Session = Depends(get_db),
     if cached:
         return cached
     
-    kpi = db.query(KPIDefinition).filter(KPIDefinition.id == kpi_id).first()
+    kpi = db.query(KPIDefinition).filter(
+        KPIDefinition.id == kpi_id,
+        KPIDefinition.entity_id == entity_id,
+    ).first()
     if not kpi:
+        # 跨企业 kpi_id 同样 404（多租户隔离 2026-09-07）
         raise HTTPException(404, "KPI不存在")
     
     values = db.query(KPIValue).filter(KPIValue.kpi_id == kpi_id).order_by(KPIValue.period.asc()).all()
@@ -422,9 +435,11 @@ async def _stream_analysis(prompt: str):
 
 
 @router.get("/dashboard-analysis-stream")
-async def dashboard_analysis_stream(role: str = Query("ceo"), db: Session = Depends(get_db)):
+async def dashboard_analysis_stream(role: str = Query("ceo"), db: Session = Depends(get_db),
+                                    entity_id: int = Depends(get_entity_id)):
     """AI分析驾驶舱数据 — SSE流式输出"""
-    cache_key = f"dashboard_analysis:{role}"
+    # 多租户隔离 2026-09-07: cache key 补 entity_id（原跨企业串缓存）
+    cache_key = f"dashboard_analysis:{role}:{entity_id}"
     cached = cache_get("ai", cache_key)
     if cached:
         # 缓存存在，直接以流的形式一次性返回
@@ -434,7 +449,10 @@ async def dashboard_analysis_stream(role: str = Query("ceo"), db: Session = Depe
             yield "data: {\"text\": \"[DONE]\"}\n\n"
         return StreamingResponse(cached_stream(), media_type="text/event-stream")
 
-    kpis = db.query(KPIDefinition).filter(KPIDefinition.status == "active").all()
+    kpis = db.query(KPIDefinition).filter(
+        KPIDefinition.entity_id == entity_id,
+        KPIDefinition.status == "active",
+    ).all()
     kpi_summary = []
     for k in kpis:
         latest = db.query(KPIValue).filter(KPIValue.kpi_id == k.id).order_by(KPIValue.period.desc()).first()
@@ -443,7 +461,10 @@ async def dashboard_analysis_stream(role: str = Query("ceo"), db: Session = Depe
             "target": k.target_value, "actual": latest.actual_value if latest else None,
             "period": latest.period if latest else None, "unit": k.unit,
         })
-    alerts = db.query(KPIAlert).filter(KPIAlert.status == "pending").count()
+    alerts = db.query(KPIAlert).filter(
+        KPIAlert.entity_id == entity_id,
+        KPIAlert.status == "pending",
+    ).count()
     kpi_text = "\n".join([f"- {k['name']}({k['code']}): 目标={k['target']}, 实际={k['actual']}({k['period']}), 维度={k['dimension']}" for k in kpi_summary if k['actual'] is not None])
     role_label = {"ceo": "CEO（总经理）", "finance": "财务部", "business": "业务部"}.get(role, "管理层")
     prompt = f"""我是一家公司的管理层，以下是当前管理会计系统的KPI数据和系统状态，请给出专业的分析和管理建议：
@@ -468,6 +489,7 @@ async def ask_question(
     request: Request,
     db: Session = Depends(get_db),
     current_user: User = Depends(require_auth),
+    entity_id: int = Depends(get_entity_id),
 ):
     """自然语言查询 — CEO问企业经营问题"""
     body = await request.json()
@@ -476,8 +498,11 @@ async def ask_question(
     if not question:
         raise HTTPException(400, "请输入问题")
     
-    # 收集系统数据作为上下文
-    kpis = db.query(KPIDefinition).filter(KPIDefinition.status == "active").all()
+    # 收集系统数据作为上下文（多租户隔离 2026-09-07: 只取本账套）
+    kpis = db.query(KPIDefinition).filter(
+        KPIDefinition.entity_id == entity_id,
+        KPIDefinition.status == "active",
+    ).all()
     kpi_context = []
     for k in kpis:
         latest = db.query(KPIValue).filter(KPIValue.kpi_id == k.id).order_by(KPIValue.period.desc()).first()
@@ -487,13 +512,21 @@ async def ask_question(
             f"{' ⚠️' + alert.alert_level if alert else ''}"
         )
     
-    # 获取改善计划
-    plans = db.query(ActionPlan).order_by(ActionPlan.created_at.desc()).limit(10).all()
+    # 获取改善计划（多租户隔离 2026-09-07: 只取本账套）
+    plans = db.query(ActionPlan).filter(
+        ActionPlan.entity_id == entity_id,
+    ).order_by(ActionPlan.created_at.desc()).limit(10).all()
     plan_context = [f"- {p.title}({p.assignee}, {p.status}, {p.progress}%)" for p in plans]
     
     # 获取预警
-    red_alerts = db.query(KPIAlert).filter(KPIAlert.status == "pending", KPIAlert.alert_level == "red").count()
-    yellow_alerts = db.query(KPIAlert).filter(KPIAlert.status == "pending", KPIAlert.alert_level == "yellow").count()
+    red_alerts = db.query(KPIAlert).filter(
+        KPIAlert.entity_id == entity_id,
+        KPIAlert.status == "pending", KPIAlert.alert_level == "red",
+    ).count()
+    yellow_alerts = db.query(KPIAlert).filter(
+        KPIAlert.entity_id == entity_id,
+        KPIAlert.status == "pending", KPIAlert.alert_level == "yellow",
+    ).count()
     
     system_context = f"""你是管理会计OS的AI助手，基于以下企业数据回答管理层问题。
 
@@ -525,9 +558,13 @@ async def ask_question(
 async def review_plans(
     db: Session = Depends(get_db),
     current_user: User = Depends(require_auth),
+    entity_id: int = Depends(get_entity_id),
 ):
     """AI复盘改善行动计划执行效果"""
-    plans = db.query(ActionPlan).order_by(ActionPlan.created_at.asc()).all()
+    # 多租户隔离 2026-09-07: ActionPlan 按 entity_id 过滤（表 2026-08-31 已加列回填）
+    plans = db.query(ActionPlan).filter(
+        ActionPlan.entity_id == entity_id,
+    ).order_by(ActionPlan.created_at.asc()).all()
     
     if not plans:
         return {"analysis": "暂无改善行动计划，无法复盘"}

backend/app/api/bot_kpis.py

@@ -6,6 +6,7 @@ from datetime import datetime
 
 from app.database import get_db
 from app.auth_middleware import require_auth
+from app.deps import get_entity_id
 from app.models import KPIDefinition, KPIValue
 from app.risk_levels import risk_level
 
@@ -142,6 +143,7 @@ def update_bot_kpi_value(
     kpi_id: int,
     data: dict,
     db: Session = Depends(get_db),
+    entity_id: int = Depends(get_entity_id),
 ):
     """更新Bot KPI当前值"""
     kpi = db.query(KPIDefinition).filter(
@@ -150,6 +152,9 @@ def update_bot_kpi_value(
     ).first()
     if not kpi:
         raise HTTPException(404, "Bot KPI不存在")
+    # 多租户隔离 2026-09-07: 归属校验，禁止跨企业修改他企 Bot KPI
+    if kpi.entity_id != entity_id:
+        raise HTTPException(404, "Bot KPI不存在")
 
     actual_value = data.get("actual_value")
     if actual_value is None:

backend/app/api/cost.py

@@ -170,6 +170,8 @@ def create_actual_cost(data: dict, db: Session = Depends(get_db),
 
 # ============================================================
 # ABC 作业成本
+# 注：ABC 作业中心/分配为单实体演示数据（ab_activities/ab_allocations 全为 entity 1，
+# 历史演示用途），OpenCode 多租户审查 2026-09-07 判定暂不强制按账套隔离；若启用多企业 ABC 需补 entity 过滤。
 # ============================================================
 
 @router.get("/abc/activities")
@@ -229,34 +231,41 @@ def list_allocations(period: Optional[str] = Query(None),
 # ============================================================
 
 @router.get("/overview")
-def cost_overview(period: Optional[str] = Query(None)):
-    """成本总览（总成本、结构占比、趋势）"""
+def cost_overview(period: Optional[str] = Query(None),
+                  db: Session = Depends(get_db),
+                  entity_id: int = Depends(get_entity_id)):
+    """成本总览（总成本、结构占比、趋势）— 多租户隔离 2026-09-07"""
     if period is None:
         period = datetime.now().strftime("%Y-%m")
-    return get_cost_overview(period)
+    return get_cost_overview(period, entity_id=entity_id)
 
 
 @router.get("/variance")
 def cost_variance(product_code: str = Query(...),
-                   period: Optional[str] = Query(None)):
-    """量差价差分析"""
+                  period: Optional[str] = Query(None),
+                  db: Session = Depends(get_db),
+                  entity_id: int = Depends(get_entity_id)):
+    """量差价差分析 — 多租户隔离 2026-09-07"""
     if period is None:
         period = datetime.now().strftime("%Y-%m")
-    return calc_product_variance(product_code, period)
+    return calc_product_variance(product_code, period, entity_id=entity_id)
 
 
 @router.get("/breakdown")
 def cost_breakdown(product_code: str = Query(...),
-                    period: Optional[str] = Query(None)):
-    """成本构成（料/工/费占比）"""
+                   period: Optional[str] = Query(None),
+                   db: Session = Depends(get_db),
+                   entity_id: int = Depends(get_entity_id)):
+    """成本构成（料/工/费占比）— 多租户隔离 2026-09-07"""
     if period is None:
         period = datetime.now().strftime("%Y-%m")
-    return get_cost_breakdown(product_code, period)
+    return get_cost_breakdown(product_code, period, entity_id=entity_id)
 
 
 @router.get("/comparison")
 def cost_method_comparison(entity: str = Query("hanke")):
-    """三种成本法对比分析：传统/变动/作业成本法 (CMA P1)"""
+    """三种成本法对比分析：传统/变动/作业成本法 (CMA P1)
+    注：纯静态教学样例数据（无 DB 访问），不涉多租户泄漏——OpenCode 审查 2026-09-07 豁免标注"""
     if entity == "hanke":
         gross_revenue = 713  # 万
         channel_rebate_rate = 0.828
@@ -317,35 +326,42 @@ def cost_method_comparison(entity: str = Query("hanke")):
 
 
 @router.get("/dashboard")
-def cost_dashboard(period: Optional[str] = Query(None)):
-    """成本分析首页—汇总数据"""
+def cost_dashboard(period: Optional[str] = Query(None),
+                   db: Session = Depends(get_db),
+                   entity_id: int = Depends(get_entity_id)):
+    """成本分析首页—汇总数据（多租户隔离 2026-09-07: ActualCost 按账套过滤）"""
     if period is None:
         period = datetime.now().strftime("%Y-%m")
-    overview = get_cost_overview(period)
+    overview = get_cost_overview(period, entity_id=entity_id)
 
-    # 获取所有产品列表
-    db = get_db().__next__()
+    # 获取所有产品列表（本账套）
+    db_session = db if db is not None else get_db().__next__()
     try:
-        products = db.query(ActualCost.product_code, ActualCost.product_name).filter(
-            ActualCost.period == period
+        products = db_session.query(ActualCost.product_code, ActualCost.product_name).filter(
+            ActualCost.entity_id == entity_id,
+            ActualCost.period == period,
         ).distinct().all()
         product_list = [{"code": p[0], "name": p[1]} for p in products]
 
         # 各产品成本
         product_costs = []
         for code, name in products:
-            costs = db.query(ActualCost).filter(
+            costs = db_session.query(ActualCost).filter(
+                ActualCost.entity_id == entity_id,
                 ActualCost.product_code == code,
                 ActualCost.period == period,
             ).all()
             total = round(sum(c.actual_cost for c in costs), 2)
             product_costs.append({"product_code": code, "product_name": name, "total_cost": total})
     finally:
-        db.close()
+        if db is None:
+            db_session.close()
 
     return {
         "period": period,
         "overview": overview,
         "products": product_list,
-        "total_cost": round(sum(p["total_cost"] for p in product_list) + overview.get("erp_cost", 0), 2) if product_list else 0,
+        # 修复既有bug(2026-09-07测试实证): 原对 product_list 求 p["total_cost"] 必 KeyError(产品列表无total_cost键)
+        # → 改为 product_costs（含各产品总额），有产品数据时端点不再 500
+        "total_cost": round(sum(p["total_cost"] for p in product_costs) + overview.get("erp_cost", 0), 2) if product_costs else 0,
     }

backend/app/api/dashboard.py

@@ -79,7 +79,10 @@ def get_dashboard_summary(role: str = Query("ceo"), period: str = Query("month")
         return cached
     kpi_total = db.query(func.count(KPIDefinition.id)).filter(
         KPIDefinition.status == "active", KPIDefinition.entity_id == entity_id).scalar()
-    alert_count = db.query(func.count(KPIAlert.id)).filter(KPIAlert.status == "pending").scalar()
+    alert_count = db.query(func.count(KPIAlert.id)).filter(
+        KPIAlert.entity_id == entity_id,
+        KPIAlert.status == "pending",
+    ).scalar()
     dims = db.query(KPIDefinition.dimension, func.count(KPIDefinition.id)).filter(
         KPIDefinition.status == "active", KPIDefinition.entity_id == entity_id
     ).group_by(KPIDefinition.dimension).all()

backend/app/api/kpis.py

@@ -731,6 +731,13 @@ def create_kpi(data: dict, db: Session = Depends(get_db), user=WRITE_ROLES, enti
         raise HTTPException(422, detail={"message": "数据校验不通过", "errors": errs})
     data["entity_id"] = entity_id  # 账套隔离: 强制写入token企业 (2026-08-23 P1a)
     data = apply_calc_type_inference(data)
+    # create 侧字段白名单（OpenCode 审查残留项修复 2026-09-07）：
+    # 仿 update 白名单 + kpi_code（新建必填），拦截 status/important_flag/主键等敏感字段注入
+    # （update 2026-08-31 已有白名单；create 此前直接 KPIDefinition(**data) 展开，可被注入）
+    _create_allow = UPDATE_KPI_WHITELIST | {"kpi_code"}
+    for k in list(data.keys()):
+        if k != "entity_id" and k not in _create_allow:
+            data.pop(k, None)
     kpi = KPIDefinition(**data)
     db.add(kpi)
     db.commit()

backend/app/utils/cost_engine.py

@@ -56,18 +56,23 @@ def calc_efficiency_variance(standard_hours: float, actual_hours: float,
 # 产品级差异分析
 # ============================================================
 
-def calc_product_variance(product_code: str, period: str) -> dict:
-    """计算指定产品在指定期间的成本差异"""
+def calc_product_variance(product_code: str, period: str, entity_id: Optional[int] = None) -> dict:
+    """计算指定产品在指定期间的成本差异（entity_id=None 时不过滤，API 调用必传——多租户隔离 2026-09-07）"""
     db = get_session_local()()
     try:
-        standards = db.query(StandardCost).filter(
+        q_std = db.query(StandardCost).filter(
             StandardCost.product_code == product_code,
             StandardCost.status == "active",
-        ).all()
-        actuals = db.query(ActualCost).filter(
+        )
+        q_act = db.query(ActualCost).filter(
             ActualCost.product_code == product_code,
             ActualCost.period == period,
-        ).all()
+        )
+        if entity_id is not None:
+            q_std = q_std.filter(StandardCost.entity_id == entity_id)
+            q_act = q_act.filter(ActualCost.entity_id == entity_id)
+        standards = q_std.all()
+        actuals = q_act.all()
 
         if not standards or not actuals:
             return {"product_code": product_code, "error": "标准成本或实际成本数据不足", "items": [], "summary": {}}
@@ -191,12 +196,16 @@ def allocate_cost(activity_id: int, period: str, product_code: str,
 # 成本总览数据
 # ============================================================
 
-def get_cost_overview(period: str) -> dict:
-    """获取成本总览数据（总成本、结构占比、趋势）"""
+def get_cost_overview(period: str, entity_id: Optional[int] = None) -> dict:
+    """获取成本总览数据（总成本、结构占比、趋势）
+    entity_id=None 时不过滤（兼容存量调用）；API 调用必须传 entity_id —— 多租户隔离 2026-09-07"""
     db = get_session_local()()
     try:
-        # 从实际成本表汇总
-        actual_costs = db.query(ActualCost).filter(ActualCost.period == period).all()
+        # 从实际成本表汇总（按账套过滤）
+        q_costs = db.query(ActualCost).filter(ActualCost.period == period)
+        if entity_id is not None:
+            q_costs = q_costs.filter(ActualCost.entity_id == entity_id)
+        actual_costs = q_costs.all()
         total_cost = sum(a.actual_cost for a in actual_costs)
 
         # 按成本类型分组
@@ -238,7 +247,10 @@ def get_cost_overview(period: str) -> dict:
 
             trend = []
             for p in reversed(months_texts):
-                costs = db.query(ActualCost).filter(ActualCost.period == p).all()
+                q_trend = db.query(ActualCost).filter(ActualCost.period == p)
+                if entity_id is not None:
+                    q_trend = q_trend.filter(ActualCost.entity_id == entity_id)
+                costs = q_trend.all()
                 total = round(sum(c.actual_cost for c in costs), 2)
                 trend.append({"period": p, "total_cost": total})
         except Exception:
@@ -256,14 +268,17 @@ def get_cost_overview(period: str) -> dict:
         db.close()
 
 
-def get_cost_breakdown(product_code: str, period: str) -> dict:
-    """获取成本构成（料/工/费占比）"""
+def get_cost_breakdown(product_code: str, period: str, entity_id: Optional[int] = None) -> dict:
+    """获取成本构成（料/工/费占比）（entity_id=None 不过滤——多租户隔离 2026-09-07，API 必传）"""
     db = get_session_local()()
     try:
-        actuals = db.query(ActualCost).filter(
+        q_actuals = db.query(ActualCost).filter(
             ActualCost.product_code == product_code,
             ActualCost.period == period,
-        ).all()
+        )
+        if entity_id is not None:
+            q_actuals = q_actuals.filter(ActualCost.entity_id == entity_id)
+        actuals = q_actuals.all()
 
         material = sum(a.actual_cost for a in actuals if a.cost_type == "material")
         labor = sum(a.actual_cost for a in actuals if a.cost_type == "labor")

backend/tests/test_multitenant_ai_20260907.py

@@ -0,0 +1,274 @@
+"""OpenCode 审查 2026-09-07 修复验证 — ai_analysis/dashboard/bot_kpis/cost/kpis 多租户隔离越权用例
+
+覆盖：
+- ai_analysis kpi_analysis 跨企业 kpi_id → 404
+- ai_analysis generate_rule_suggestions 跨企业预警不进建议
+- ai_analysis review-plans / ask 按账套过滤（统计口径 + 不 500）
+- dashboard summary alert_count 按账套
+- bot_kpis update 跨企业写 → 404 / 同企业写 → 200
+- cost dashboard/overview 按账套（ActualCost 多企业共用确认）
+- kpis create 白名单：status/important_flag 注入被忽略
+"""
+import pytest
+from fastapi.testclient import TestClient
+from sqlalchemy.orm import Session
+
+from app.models import KPIDefinition, KPIValue, KPIAlert, ActionPlan, AISuggestion, Entity, UserEntity, ActualCost
+from app.api import ai_analysis
+from tests.conftest import create_test_kpi, create_test_user, get_token_for_user, auth_header
+
+
+def _ensure_entity(db: Session, eid: int) -> Entity:
+    ent = db.query(Entity).filter(Entity.id == eid).first()
+    if not ent:
+        ent = Entity(id=eid, name=f"企业{eid}", short_name=f"E{eid}", status="active")
+        db.add(ent)
+        db.commit()
+        db.refresh(ent)
+    return ent
+
+
+def _seed_kpi(db: Session, entity_id: int, code: str) -> KPIDefinition:
+    return create_test_kpi(db, kpi_code=code, entity_id=entity_id,
+                           dimension="finance", status="active", target_value=100.0)
+
+
+def _seed_user(db: Session, entity_id: int):
+    """默认测试用户，登录指定账套"""
+    create_test_user(db)
+    _ensure_entity(db, entity_id)
+    # 无授权记录 → login 会默认授予全部 active 企业（_ensure_default_grants）
+    db.commit()
+
+
+def _token(client: TestClient, entity_id: int) -> str:
+    """登录并返回指定账套的 token（get_token_for_user 固定 entity1，这里手动登录）"""
+    resp = client.post("/api/cma/auth/login", json={
+        "username": "testadmin", "password": "admin123", "entity_id": entity_id,
+    })
+    assert resp.status_code == 200, f"登录 entity{entity_id} 失败: {resp.text[:300]}"
+    return resp.json().get("token")
+
+
+# ============================================================
+# ai_analysis
+# ============================================================
+class TestAiAnalysisIsolation:
+    def test_kpi_analysis_cross_entity_404(self, client: TestClient, db: Session):
+        """跨企业 kpi_id 调 kpi-analysis → 404（不泄漏他企 KPI 数据）"""
+        _seed_user(db, 1)
+        _ensure_entity(db, 2)
+        k_e2 = _seed_kpi(db, entity_id=2, code="MT_AI_E2")
+        db.commit()
+        token1 = _token(client, 1)
+
+        r = client.get(f"/api/cma/ai/kpi-analysis/{k_e2.id}", headers=auth_header(token1))
+        assert r.status_code == 404, f"跨企业 kpi 应 404: {r.text[:200]}"
+
+    def test_kpi_analysis_own_entity_ok(self, client: TestClient, db: Session):
+        """本企业 kpi_id → 200（回归保护）"""
+        _seed_user(db, 1)
+        k_e1 = _seed_kpi(db, entity_id=1, code="MT_AI_E1")
+        db.commit()
+        token1 = _token(client, 1)
+
+        r = client.get(f"/api/cma/ai/kpi-analysis/{k_e1.id}", headers=auth_header(token1))
+        assert r.status_code == 200, r.text[:300]
+
+    def test_generate_suggestions_cross_entity_alert_excluded(self, db: Session):
+        """generate_rule_suggestions(entity=1) 不消费 entity2 的 pending 预警"""
+        _ensure_entity(db, 2)
+        k1 = _seed_kpi(db, entity_id=1, code="MT_SUG_E1")
+        k2 = _seed_kpi(db, entity_id=2, code="MT_SUG_E2")
+        db.add(KPIAlert(kpi_id=k1.id, alert_level="red", alert_message="e1预警", status="pending", entity_id=1))
+        a2 = KPIAlert(kpi_id=k2.id, alert_level="red", alert_message="e2预警", status="pending", entity_id=2)
+        db.add(a2)
+        db.commit()
+
+        ai_analysis.generate_rule_suggestions(db, entity_id=1, source="test", user_id=None)
+        db.commit()
+
+        # entity1 的预警 → 生成 alert 类建议
+        sugs = db.query(AISuggestion).filter(AISuggestion.source == "test").all()
+        alert_sugs = [s for s in sugs if s.suggestion_type == "action_plan" and s.target_type == "alert"]
+        assert len(alert_sugs) >= 1, "本账套 pending 预警应生成建议"
+        # 无一建议引用 entity2 的预警 id / KPI
+        sug_alert_ids = {s.suggestion_data.get("alert_id") for s in alert_sugs}
+        sug_kpi_ids = {s.suggestion_data.get("kpi_id") for s in alert_sugs}
+        assert a2.id not in sug_alert_ids, "跨企业预警不应进入建议"
+        assert k2.id not in sug_kpi_ids, "跨企业 KPI 不应进入建议"
+
+    def test_review_plans_entity_scoped(self, client: TestClient, db: Session):
+        """review-plans 只复盘本账套 ActionPlan（stats.total 口径隔离）"""
+        _seed_user(db, 1)
+        _ensure_entity(db, 2)
+        k1 = _seed_kpi(db, entity_id=1, code="MT_PL_E1")
+        k2 = _seed_kpi(db, entity_id=2, code="MT_PL_E2")
+        db.add(ActionPlan(entity_id=1, alert_id=None, kpi_id=k1.id, title="e1计划", assignee="张三",
+                          priority="high", status="in_progress", progress=50, created_by="系统"))
+        db.add(ActionPlan(entity_id=2, alert_id=None, kpi_id=k2.id, title="e2计划", assignee="李四",
+                          priority="high", status="pending", progress=0, created_by="系统"))
+        db.commit()
+        token1 = _token(client, 1)
+
+        r = client.post("/api/cma/ai/review-plans", headers=auth_header(token1))
+        assert r.status_code == 200, r.text[:300]
+        stats = r.json()["stats"]
+        assert stats["total"] == 1, f"应只统计本账套1条，实际 {stats}"
+
+    def test_ask_question_no_crash_cross_entity(self, client: TestClient, db: Session):
+        """ask 上下文含他企 KPI 时不 500 且正常返回（entity 过滤后无泄漏）"""
+        _seed_user(db, 1)
+        _ensure_entity(db, 2)
+        k2 = _seed_kpi(db, entity_id=2, code="MT_ASK_E2")
+        db.add(KPIValue(kpi_id=k2.id, period="2026-06", actual_value=88.0, entity_id=2))
+        db.commit()
+        token1 = _token(client, 1)
+
+        r = client.post("/api/cma/ai/ask", headers=auth_header(token1),
+                        json={"question": "企业营收情况如何"})
+        assert r.status_code == 200, r.text[:300]
+        assert "answer" in r.json()
+
+
+# ============================================================
+# dashboard summary alert_count
+# ============================================================
+class TestDashboardAlertCountIsolation:
+    def test_alert_count_per_entity(self, client: TestClient, db: Session):
+        """summary alert_count 按账套：entity1 只数自己的 pending 预警"""
+        from app.utils import cache as cache_util
+        cache_util.delete("dashboard")  # 防跨测试缓存污染
+
+        _seed_user(db, 1)
+        _ensure_entity(db, 2)
+        k1 = _seed_kpi(db, entity_id=1, code="MT_DB_E1")
+        k2 = _seed_kpi(db, entity_id=2, code="MT_DB_E2")
+        db.add(KPIAlert(kpi_id=k1.id, alert_level="red", alert_message="e1", status="pending", entity_id=1))
+        db.add(KPIAlert(kpi_id=k2.id, alert_level="red", alert_message="e2a", status="pending", entity_id=2))
+        db.add(KPIAlert(kpi_id=k2.id, alert_level="yellow", alert_message="e2b", status="resolved", entity_id=2))
+        db.commit()
+        token1 = _token(client, 1)
+        token2 = _token(client, 2)
+
+        r1 = client.get("/api/cma/dashboard/summary", headers=auth_header(token1))
+        assert r1.status_code == 200, r1.text[:200]
+        assert r1.json()["alert_count"] == 1, f"entity1 应只数自己的1条: {r1.json()['alert_count']}"
+
+        cache_util.delete("dashboard")
+        r2 = client.get("/api/cma/dashboard/summary", headers=auth_header(token2))
+        assert r2.status_code == 200, r2.text[:200]
+        assert r2.json()["alert_count"] == 1, f"entity2 应只数自己的 pending 1条(不含resolved): {r2.json()['alert_count']}"
+
+
+# ============================================================
+# bot_kpis 归属校验
+# ============================================================
+class TestBotKpisOwnership:
+    def _bot_kpi(self, db: Session, entity_id: int, code: str) -> KPIDefinition:
+        return create_test_kpi(db, kpi_code=code, entity_id=entity_id, status="active",
+                               bot_source="finance-bot", kpi_name=code, target_value=100.0)
+
+    def test_cross_entity_update_404(self, client: TestClient, db: Session):
+        """entity1 用户改 entity2 的 Bot KPI → 404"""
+        _seed_user(db, 1)
+        _ensure_entity(db, 2)
+        k2 = self._bot_kpi(db, entity_id=2, code="MT_BOT_E2")
+        db.commit()
+        token1 = _token(client, 1)
+
+        r = client.post(f"/api/cma/bot-kpis/{k2.id}/value", headers=auth_header(token1),
+                        json={"actual_value": 55, "period": "2026-09"})
+        assert r.status_code == 404, f"跨企业改 Bot KPI 应 404: {r.status_code} {r.text[:200]}"
+
+    def test_same_entity_update_ok(self, client: TestClient, db: Session):
+        """本企业改自己的 Bot KPI → 200（回归保护）"""
+        _seed_user(db, 1)
+        k1 = self._bot_kpi(db, entity_id=1, code="MT_BOT_E1")
+        db.commit()
+        token1 = _token(client, 1)
+
+        r = client.post(f"/api/cma/bot-kpis/{k1.id}/value", headers=auth_header(token1),
+                        json={"actual_value": 66, "period": "2026-09"})
+        assert r.status_code == 200, r.text[:300]
+
+
+# ============================================================
+# cost 多账套（ActualCost 共用表确认 → 过滤）
+# ============================================================
+class TestCostIsolation:
+    def _seed_actual(self, db: Session, entity_id: int, code: str, amount: float, period: str = "2026-06"):
+        ac = ActualCost(entity_id=entity_id, period=period, product_code=code, product_name=code,
+                        cost_type="material", item_name="材料", actual_quantity=1.0,
+                        actual_price=amount, actual_cost=amount, source="manual")
+        db.add(ac)
+
+    def test_cost_dashboard_entity_filtered(self, client: TestClient, db: Session):
+        """cost/dashboard 产品列表/总额按账套过滤（ActualCost 多企业共用）"""
+        _seed_user(db, 1)
+        _ensure_entity(db, 2)
+        self._seed_actual(db, 1, "PROD_E1", 100.0)
+        self._seed_actual(db, 2, "PROD_E2", 200.0)
+        db.commit()
+        token1 = _token(client, 1)
+        token2 = _token(client, 2)
+
+        r1 = client.get("/api/cma/cost/dashboard?period=2026-06", headers=auth_header(token1))
+        assert r1.status_code == 200, r1.text[:300]
+        codes1 = {p["code"] for p in r1.json()["products"]}
+        assert codes1 == {"PROD_E1"}, f"entity1 不应看到 entity2 产品: {codes1}"
+
+        r2 = client.get("/api/cma/cost/dashboard?period=2026-06", headers=auth_header(token2))
+        assert r2.status_code == 200, r2.text[:300]
+        codes2 = {p["code"] for p in r2.json()["products"]}
+        assert codes2 == {"PROD_E2"}, f"entity2 不应看到 entity1 产品: {codes2}"
+
+    def test_cost_overview_entity_filtered(self, client: TestClient, db: Session):
+        """cost/overview manual_cost 按账套过滤"""
+        _seed_user(db, 1)
+        _ensure_entity(db, 2)
+        self._seed_actual(db, 1, "OV_E1", 100.0)
+        self._seed_actual(db, 2, "OV_E2", 300.0)
+        db.commit()
+        token1 = _token(client, 1)
+        token2 = _token(client, 2)
+
+        r1 = client.get("/api/cma/cost/overview?period=2026-06", headers=auth_header(token1))
+        assert r1.status_code == 200, r1.text[:300]
+        assert r1.json()["manual_cost"] == 100.0, f"entity1 manual_cost 应100: {r1.json()}"
+
+        r2 = client.get("/api/cma/cost/overview?period=2026-06", headers=auth_header(token2))
+        assert r2.status_code == 200, r2.text[:300]
+        assert r2.json()["manual_cost"] == 300.0, f"entity2 manual_cost 应300: {r2.json()}"
+
+
+# ============================================================
+# kpis create 白名单（mass-assignment 残留项）
+# ============================================================
+class TestKpiCreateWhitelist:
+    def test_create_injected_sensitive_fields_ignored(self, client: TestClient, db: Session):
+        """create 带 status/important_flag 注入 → 被忽略，落库为默认值"""
+        _seed_user(db, 1)
+        token1 = _token(client, 1)
+
+        resp = client.post("/api/cma/kpis", headers=auth_header(token1), json={
+            "kpi_code": "F_MT_INJ_001",
+            "kpi_name": "注入测试指标",
+            "dimension": "finance",
+            "target_value": 100.0,
+            "unit": "%",
+            "formula": "测试公式",
+            "data_source": "测试系统",
+            "data_owner": "测试管理员",
+            "status": "disabled",      # 敏感字段：注入应被忽略 → 默认 active
+            "important_flag": 1,       # 敏感字段：注入应被忽略 → 默认 0
+            "entity_id": 999,          # 归属字段：强制覆盖为 token 企业 1
+        })
+        assert resp.status_code == 200, resp.text[:300]
+        data = resp.json()
+        assert data["status"] == "active", f"status 注入应被忽略: {data.get('status')}"
+        assert data.get("important_flag", 0) == 0, f"important_flag 注入应被忽略: {data.get('important_flag')}"
+
+        row = db.query(KPIDefinition).filter(KPIDefinition.kpi_code == "F_MT_INJ_001").first()
+        assert row is not None
+        assert row.entity_id == 1, f"entity_id 应强制为 token 企业1: {row.entity_id}"