diff --git a/backend/scripts/verify_bot_write_tiering_e2e.py b/backend/scripts/verify_bot_write_tiering_e2e.py
new file mode 100644
index 00000000..68d2bf39
--- /dev/null
+++ b/backend/scripts/verify_bot_write_tiering_e2e.py
@@ -0,0 +1,219 @@
+#!/usr/bin/env python3
+"""生产E2E实证: Bot写操作分级+分层授权 (P0-② 2026-09-03)
+
+链路验证:
+  A. Bot通道(X-BOT-KEY)创建 kpi_target 高危建议 → status=unapplied
+  B. 人token apply → KPI目标值变更 + apply_detail(before→after) + applied_by/applied_at
+  C. 低危直写 bot-bridge /kpi-result (X-BRIDGE-TOKEN) → KPIValue.source_batch 落库
+  D. bot_audit.log 留痕(两条: 建议创建L2 + kpi-result L2)
+
+数据纪律: 临时KPI/建议/值 全部带【系统验证临时数据-验证后删除】标记,
+验证完成后精确清理并回查 COUNT=0。审计日志文件行保留(留痕证据)。
+
+用法: cd /root/cma-management/backend && venv/bin/python3 scripts/verify_bot_write_tiering_e2e.py
+"""
+import json
+import os
+import sys
+import urllib.request
+import urllib.error
+
+BASE = "http://127.0.0.1:8010"
+MARK = "【系统验证临时数据-验证后删除】"
+KPI_CODE = "E2E_WT_TIER"
+SUG_PREFIX = "E2E-验证"
+CUR_PERIOD = "2026-09"
+BASELINE_TARGET = 100.0
+NEW_TARGET = 888.88
+LOW_VALUE = 777.0
+SOURCE = "e2e-bot-tier-20260903"
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+os.environ.setdefault("CMA_DB_HOST", "127.0.0.1")
+from app.database import get_db  # noqa: E402
+from app.models import (  # noqa: E402
+    BotBridgeConfig, KPIDefinition, KPIValue, AISuggestion, OperationLog, Entity,
+)
+
+
+def http(method, path, body=None, headers=None):
+    data = json.dumps(body).encode() if body is not None else None
+    req = urllib.request.Request(BASE + path, data=data, method=method)
+    req.add_header("Content-Type", "application/json")
+    for k, v in (headers or {}).items():
+        req.add_header(k, v)
+    try:
+        with urllib.request.urlopen(req, timeout=30) as r:
+            return r.status, json.loads(r.read().decode())
+    except urllib.error.HTTPError as e:
+        return e.code, json.loads(e.read().decode())
+
+
+def main():
+    failures = []
+
+    def check(name, cond, detail=""):
+        tag = "PASS" if cond else "FAIL"
+        print(f"  [{tag}] {name}" + (f" | {detail}" if detail else ""))
+        if not cond:
+            failures.append(name)
+
+    # ── 0. 取Bot密钥与bridge token ──
+    with open(os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), ".env")) as f:
+        env = {}
+        for line in f:
+            if "=" in line and not line.strip().startswith("#"):
+                k, _, v = line.strip().partition("=")
+                env[k] = v.strip().strip('"').strip("'")
+    keys = json.loads(env["CMA_BOT_API_KEYS"])
+    fin_key = next(k for k, v in keys.items() if v.get("name") == "财务BOT")
+
+    db = next(get_db())
+    bridge = db.query(BotBridgeConfig).filter_by(is_active=True).first()
+    bridge_token = bridge.token if bridge else None
+    db.close()
+
+    # ── 1. 登录(人token, entity=1) ──
+    st, r = http("POST", "/api/cma/auth/login",
+                 {"username": "admin", "password": "admin123", "entity_id": 1})
+    check("登录 admin entity=1", st == 200, f"http={st}")
+    token = r.get("token") if st == 200 else ""
+    auth = {"Authorization": f"Bearer {token}"}
+
+    # ── 2. 临时KPI准备(存在则复用, 目标值重置基线) ──
+    db = next(get_db())
+    db.rollback()
+    kpi = db.query(KPIDefinition).filter(
+        KPIDefinition.kpi_code == KPI_CODE,
+        KPIDefinition.entity_id == 1,
+    ).first()
+    if not kpi:
+        kpi = KPIDefinition(
+            entity_id=1, kpi_code=KPI_CODE,
+            kpi_name=f"{MARK}写分层E2E临时KPI",
+            dimension="finance", category="financial_report", formula="-",
+            data_source_type="manual", data_source=f"{MARK}E2E", data_owner="系统验证",
+            frequency="monthly", unit="万元", target_value=BASELINE_TARGET,
+            kpi_level="operational", status="active",
+        )
+        db.add(kpi)
+        db.commit()
+        db.refresh(kpi)
+        kpi_id = kpi.id
+        db.close()
+    else:
+        kpi_id = kpi.id
+        kpi.target_value = BASELINE_TARGET
+        db.commit()
+        db.refresh(kpi)
+        db.close()
+    print(f"  临时KPI id={kpi_id} code={KPI_CODE} baseline={BASELINE_TARGET}")
+
+    # ── 3. Bot创建高危建议(kpi_target) ──
+    sug_title = f"{SUG_PREFIX} 上调临时KPI目标→{NEW_TARGET}"
+    st, r = http("POST", "/api/cma/bot/ai/suggestions", {
+        "suggestion_type": "kpi_target", "title": sug_title, "content": "分层授权E2E实证",
+        "source": "bot", "entity_id": 1, "target_id": kpi_id,
+        "suggestion_data": {"kpi_id": kpi_id, "target_value": NEW_TARGET},
+    }, {"X-BOT-KEY": fin_key})
+    check("A1 Bot创建建议 200", st == 200, f"http={st} body={str(r)[:200]}")
+    sug_id = r.get("data", {}).get("id") if st == 200 else None
+    check("A2 status=unapplied", st == 200 and r["data"]["status"] == "unapplied",
+          f"id={sug_id} source={r.get('data', {}).get('source')}")
+
+    # ── 4. 人apply ──
+    st, r = http("POST", f"/api/cma/ai/suggestions/{sug_id}/apply",
+                 {"action": "kpi_target", "target_value": NEW_TARGET}, auth)
+    check("B1 apply 200", st == 200, f"http={st} body={str(r)[:200]}")
+    ad = r.get("data", {}) if st == 200 else {}
+    detail = (ad.get("apply_detail") or [{}])[0]
+    check("B2 apply_detail before→after", detail.get("before") == BASELINE_TARGET
+          and detail.get("after") == NEW_TARGET, str(detail)[:160])
+    check("B3 applied_by/applied_at", bool(ad.get("applied_by")) and bool(ad.get("applied_at")),
+          f"by={ad.get('applied_by')} at={ad.get('applied_at')}")
+
+    db = next(get_db())
+    db.rollback()
+    kpi2 = db.query(KPIDefinition).filter(KPIDefinition.id == kpi_id).first()
+    db_target = getattr(kpi2, "target_value", None) if kpi2 else None
+    check("B4 KPI目标值生效", db_target == NEW_TARGET,
+          f"db_target={db_target}")
+    db.close()
+
+    # ── 5. 低危直写 /kpi-result ──
+    st, r = http("POST", "/api/cma/bot-bridge/kpi-result", {
+        "entity_id": 1, "kpi_code": KPI_CODE, "period": CUR_PERIOD,
+        "value": LOW_VALUE, "source": SOURCE, "remark": MARK,
+    }, {"X-BRIDGE-TOKEN": bridge_token})
+    check("C1 kpi-result 200", st == 200, f"http={st} body={str(r)[:200]}")
+
+    db = next(get_db())
+    db.rollback()
+    val = db.query(KPIValue).filter(
+        KPIValue.kpi_id == kpi_id, KPIValue.period == CUR_PERIOD,
+    ).order_by(KPIValue.id.desc()).first()
+    check("C2 source_batch落库", val is not None and val.source_batch == SOURCE,
+          f"source_batch={val.source_batch if val else None} actual={val.actual_value if val else None}")
+    db.close()
+
+    # ── 6. audit留痕 ──
+    audit_path = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
+                              "logs", "bot_audit.log")
+    lines = []
+    if os.path.exists(audit_path):
+        with open(audit_path, encoding="utf-8") as f:
+            lines = f.readlines()
+    audit_str = "".join(lines[-60:])
+    check("D1 audit含建议创建记录",
+          "/api/cma/bot/ai/suggestions" in audit_str and '"status": 200' in audit_str,
+          f"sug_id={sug_id}")
+    check("D2 audit含kpi-result记录",
+          "/api/cma/bot-bridge/kpi-result" in audit_str and '"status": 200' in audit_str)
+    # 取最近一条(倒序)建议创建成功记录验证字段；早于重启的401是pytest进程写文件的噪音
+    rec = None
+    for l in reversed(lines[-60:]):
+        try:
+            r = json.loads(l)
+        except Exception:
+            continue
+        if r.get("endpoint") == "/api/cma/bot/ai/suggestions" and r.get("status") == 200:
+            rec = r
+            break
+    if rec:
+        check("D3 建议创建audit字段(bot_name/L2)", rec.get("bot_name") == "财务BOT"
+              and rec.get("risk_level") == "L2", str(rec)[:200])
+
+    # ── 7. 清理(标记精确删除) + 回查 ──
+    db = next(get_db())
+    db.rollback()
+    del_sug = db.query(AISuggestion).filter(
+        AISuggestion.title.like(f"{SUG_PREFIX}%"), AISuggestion.title.like("%上调临时KPI目标%"),
+    ).delete(synchronize_session=False)
+    del_val = db.query(KPIValue).filter(
+        KPIValue.kpi_id == kpi_id, KPIValue.source_batch == SOURCE,
+    ).delete(synchronize_session=False)
+    db.query(OperationLog).filter(OperationLog.action == "ai_suggestion_apply",
+                                  OperationLog.target_id == kpi_id).delete(synchronize_session=False)
+    del_kpi = db.query(KPIDefinition).filter(KPIDefinition.id == kpi_id).delete(synchronize_session=False)
+    db.commit()
+    db.rollback()
+    left_sug = db.query(AISuggestion).filter(
+        AISuggestion.title.like(f"{SUG_PREFIX}%"), AISuggestion.title.like("%上调临时KPI目标%"),
+    ).count()
+    left_val = db.query(KPIValue).filter(
+        KPIValue.kpi_id == kpi_id, KPIValue.source_batch == SOURCE,
+    ).count()
+    left_kpi = db.query(KPIDefinition).filter(KPIDefinition.id == kpi_id).count()
+    check("E1 清理回查=0", left_sug == 0 and left_val == 0 and left_kpi == 0,
+          f"sug={left_sug} val={left_val} kpi={left_kpi}")
+    db.close()
+
+    print()
+    if failures:
+        print(f"E2E RESULT: FAIL ({len(failures)}): {failures}")
+        sys.exit(1)
+    print(f"E2E RESULT: PASS | 建议id={sug_id}(已清理) kpi_id={kpi_id}(已清理) 审计行保留于 logs/bot_audit.log")
+
+
+if __name__ == "__main__":
+    main()
diff --git a/docs/bot-write-tiering-20260903.md b/docs/bot-write-tiering-20260903.md
index 9ad120f6..8fbef4fc 100644
--- a/docs/bot-write-tiering-20260903.md
+++ b/docs/bot-write-tiering-20260903.md
@@ -67,4 +67,8 @@
 
 ## 七、验收记录（E2E/回归）
 
-见任务完成回报与 backend/tests/test_bot_suggestion_channel.py（新增 pytest 回归覆盖）。
+- 回归: backend/tests/test_bot_suggestion_channel.py（12条：鉴权/语义/多租户/幂等/审批闭环/审计/收口）；test_risk_levels.py 计数更新 L2=6。
+- 生产E2E: backend/scripts/verify_bot_write_tiering_e2e.py（可复跑，标记+清理+回查）2026-09-03 全 PASS：
+  A Bot创建kpi_target建议→unapplied → B 人apply→KPI目标100→888.88 + apply_detail(before/after) + applied_by=管理员/applied_at → C /kpi-result 直写 source_batch=e2e-bot-tier-20260903 落库 → D bot_audit.log 两条 L2 记录（财务BOT / bot-bridge）→ E 清理回查=0。
+- pytest 全量: 726 passed / 4 failed(日历漂移预存：budget roll_forward + ai_suggestions budget_overrun + roadmap budget_headroom×2，非本次回归) / 1 xfailed。
+- 后端重启: systemctl restart cma-backend（ActiveEnter 晚于 commit）; /health ok。