feat(simulator): 增加受控压力测试能力与实体授权探测

tools/integration-simulator/app.py

@@ -16,6 +16,7 @@ import json
 import os
 import socket
 import sys
+import threading
 import time
 import uuid
 from pathlib import Path
@@ -34,6 +35,9 @@ from catalog import sample_adapters as sa
 from clients import aidop_probe
 from clients.inbound_client import InboundClient
 from clients.redaction import mask_access_key, redact_obj
+from load import controls as load_controls
+from load import metrics as load_metrics
+from load import runner_p3
 from providers import fault_catalog, mock_api, mock_db
 
 DEFAULT_CONTROL_PORT = 8017
@@ -43,6 +47,10 @@ MAX_RUNS = 200
 CONFIG_STATES = ("READY", "CONFIG_NOT_REGISTERED", "CONFIG_NOT_ENABLED",
                  "GRANT_MISSING", "SOURCE_UNREACHABLE", "RUN_FAILED")
 
+# 入站实体授权探测缓存（60s TTL；探测用 GET schema，无副作用）
+_AUTH_PROBE_CACHE: dict = {"at": 0.0, "data": None}
+_AUTH_PROBE_TTL = 60.0
+
 
 def _env(name: str) -> str:
     return os.environ.get(name, "") or ""
@@ -78,6 +86,13 @@ CONFIG = load_config()
 MOCK_STATE = mock_api.MockApiState()
 MOCK_APP = mock_api.create_app(MOCK_STATE)
 
+# ---- 第三阶段压力测试全局（默认关闭，重启即清）----
+LOAD_GUARD = load_controls.LoadGuard()
+LOAD_METRICS = load_metrics.LoadMetrics()
+LOAD_RUNNER: runner_p3.P3Runner | None = None
+LOAD_REPORTS = None   # 在 ReportStore 定义后创建
+DB_GEN = {"thread": None, "result": None, "error": ""}
+
 
 class RunStore:
     def __init__(self, limit: int = MAX_RUNS):
@@ -257,6 +272,65 @@ class FaultReportPayload(BaseModel):
     note: str = ""
 
 
+class LoadPlanPayload(BaseModel):
+    """压力测试计划（§16 十项 + §17 参数 + 通道专属参数，§20 安全限制）。
+
+    校验由 load.controls.validate_plan 执行；本模型只负责字段承接。
+    """
+    channel: str                            # DB_SYNC | API_PULL | API_INBOUND
+    objectCode: str = ""
+    entityCode: str = ""
+    runId: str | None = None
+    # §16 执行前必填十项
+    environment: str = ""
+    tenant: str = ""
+    owner: str = ""
+    timeWindow: str = ""
+    maxDataRows: int = 0
+    maxConcurrency: int = 0
+    maxQps: int = 0
+    maxDurationSeconds: int = 0
+    stopThreshold: str = ""
+    cleanupPlan: str = ""
+    # 安全
+    confirmed: bool = False                 # 共享环境负责人确认
+    escalationConfirm: bool = False         # 提高上限二次确认
+    # §17 通用参数
+    warmupSeconds: int = 0
+    batchSize: int = 1
+    requestIntervalMs: int = 0
+    timeoutSeconds: int = 30
+    maxRetries: int = 0
+    sampleMode: str = "fixed"               # fixed | random
+    maxConsecutiveErrors: int = 0
+    maxTimeouts: int = 0
+    maxRateLimited: int = 0
+    errorRate: float = 0.0                  # P2 Mock 随机 500 比例 / P3 错误数据比例
+    # P1 专属（§18）
+    size: str = ""                          # S | M | L | XL（记录档位，行数以 maxDataRows 为准）
+    incrementRatio: float = 1.0
+    fieldWidth: int = 0
+    # P2 专属（§19）
+    pageSize: int = 0
+    totalRowsLimit: int = 0
+    randomDelay: float = 0.0
+    rateLimit: float = 0.0
+    # P3 专属（§20）
+    mode: str = "push"                      # push | bulk
+    rampSteps: list[dict] | None = None     # 阶梯升压 [{duration,qps,concurrency}]
+    idemReplay: bool = False                # 幂等重复
+
+
+class LoadReportPayload(BaseModel):
+    """压力报告人工观察部分（自动指标与计划由后端快照填充，§18 验收/§17 末句）。"""
+    runId: str = ""
+    channel: str = ""
+    aidopResource: str = ""     # Ai-DOP CPU/内存/队列/数据库人工观察结果
+    aidopFlow: str = ""         # Ai-DOP 同步/入站流水核对结果
+    conclusion: str = ""
+    notes: str = ""
+
+
 CHANNELS = ("DB_SYNC", "API_PULL", "API_INBOUND")
 CHANNEL_LABELS = {
     "DB_SYNC": "方式甲 DB_SYNC（数据库同步）",
@@ -339,6 +413,9 @@ def build_report_markdown(entry: dict) -> str:
 
 REPORTS = ReportStore()
 
+# 压力报告复用 ReportStore（保存前脱敏、内存态不落盘）
+LOAD_REPORTS = ReportStore(limit=20)
+
 
 class FaultReportStore(ReportStore):
     """异常测试报告存储（任务书 §15 N4）。复用脱敏与内存态约定。
@@ -743,6 +820,52 @@ def precheck_inbound_one(entity_code: str):
 # API_INBOUND 执行（P1-E）
 # ---------------------------------------------------------------------------
 
+@app.get("/api/inbound/auth-probe")
+def inbound_auth_probe(refresh: int = 0):
+    """探测全部入站实体在 Ai-DOP 侧的授权状态（GET schema，只读无副作用）。
+
+    Secret 不离开后端进程；结果缓存 60 秒，refresh=1 强制重测。
+    前端实体下拉据此只展示已授权实体，避免选到 403 未授权实体。
+    """
+    global _AUTH_PROBE_CACHE
+    now = time.monotonic()
+    if not refresh and _AUTH_PROBE_CACHE["data"] is not None \
+            and now - _AUTH_PROBE_CACHE["at"] < _AUTH_PROBE_TTL:
+        return {"ok": True, "cached": True, **_AUTH_PROBE_CACHE["data"]}
+    reach = aidop_probe.check_reachable(CONFIG["aidopBaseUrl"], CONFIG["allowedTargetHosts"],
+                                        timeout=5)
+    if not reach.get("reachable"):
+        # Ai-DOP 不可达：不逐个探测（避免 26 次连接等待），全部如实标记为不可用
+        entities = {inbound["entityCode"]: False
+                    for obj in sa.load_catalog()["objects"]
+                    if (inbound := ((obj or {}).get("apiInbound") or {})) and inbound.get("supported")}
+        data = {"entities": entities,
+                "probedAt": time.strftime("%Y-%m-%d %H:%M:%S"),
+                "probeErrors": [f"Ai-DOP 不可达: {reach.get('error')}"]}
+        _AUTH_PROBE_CACHE.update(at=time.monotonic(), data=data)
+        return {"ok": True, "cached": False, **data}
+    client = _build_inbound_client()
+    entities: dict[str, bool] = {}
+    probe_errors: list[str] = []
+    for obj in sa.load_catalog()["objects"]:
+        inbound = (obj or {}).get("apiInbound") or {}
+        if not inbound.get("supported"):
+            continue
+        entity = inbound["entityCode"]
+        try:
+            status, _, _ = client.op_schema(entity,
+                                            contract_version=inbound.get("contractVersion"))
+            entities[entity] = status == 200
+            if status not in (200, 403):
+                probe_errors.append(f"{entity}: HTTP {status}")
+        except Exception as ex:  # noqa: BLE001 —— 单实体失败不阻断整体探测
+            probe_errors.append(f"{entity}: {type(ex).__name__}")
+    data = {"entities": entities, "probedAt": time.strftime("%Y-%m-%d %H:%M:%S"),
+            "probeErrors": probe_errors}
+    _AUTH_PROBE_CACHE.update(at=time.monotonic(), data=data)
+    return {"ok": True, "cached": False, **data}
+
+
 @app.post("/api/inbound/exec")
 def inbound_exec(payload: InboundExecPayload):
     obj, entity = _resolve_entity(payload)
@@ -1209,6 +1332,307 @@ def fault_report_delete(report_id: str):
     return {"ok": True}
 
 
+# ---------------------------------------------------------------------------
+# 压力测试（第三阶段 §16-§21）
+# ---------------------------------------------------------------------------
+
+def _apply_p2_load(plan: dict) -> dict:
+    """P2：把压力参数注入 Mock API（§19 配置每次返回行数/总量/延迟/错误比例/限流阈值）。"""
+    s = MOCK_STATE.settings
+    s.load_mode = True
+    s.page_size = int(plan.get("pageSize") or 0)
+    s.total_rows_limit = int(plan.get("totalRowsLimit") or 0)
+    s.random_delay = float(plan.get("randomDelay") or 0)
+    s.error_rate = float(plan.get("errorRate") or 0)
+    s.rate_limit = float(plan.get("rateLimit") or 0)
+    MOCK_STATE.load_stats.reset()
+    return {"ok": True, "settings": mock_api_settings(), "mockRunning": _mock_running(),
+            "note": "" if _mock_running() else "提示：Mock API 未启动，先点「启动 Mock API」"}
+
+
+def _stop_p2_load() -> dict:
+    """P2：实时停止压力模式，恢复正常模式（§19）。"""
+    s = MOCK_STATE.settings
+    s.load_mode = False
+    s.page_size = 0
+    s.total_rows_limit = 0
+    s.random_delay = 0.0
+    s.error_rate = 0.0
+    s.rate_limit = 0.0
+    return {"ok": True, "note": "Mock 压力模式已停止，恢复正常模式"}
+
+
+def _run_p1_generation(obj: dict, plan: dict, run_id: str) -> None:
+    """P1 后台线程：批量生成档位数据（§18 批量生成/耗时行数/增量比例/字段宽度/一键停止）。"""
+    DB_GEN.update({"result": None, "error": ""})
+    table = (obj.get("dbSync") or {}).get("table")
+    rows = sa.adapt_db_rows(obj, run_id)
+    started = time.monotonic()
+    try:
+        result = mock_db.bulk_generate(
+            table, rows, int(plan["maxDataRows"]),
+            increment_ratio=float(plan.get("incrementRatio") or 1.0),
+            field_width=int(plan.get("fieldWidth") or 0),
+            batch_size=1000, stop_event=LOAD_GUARD.stop)
+        LOAD_METRICS.mark_start()
+        LOAD_METRICS.record(200, time.monotonic() - started,
+                            rows=result["loaded"], batches=result["batches"])
+        LOAD_METRICS.mark_finish()
+        DB_GEN["result"] = result
+    except Exception as ex:  # noqa: BLE001
+        LOAD_METRICS.mark_start()
+        LOAD_METRICS.record(500, time.monotonic() - started)
+        LOAD_METRICS.mark_finish()
+        DB_GEN["error"] = f"{type(ex).__name__}: {ex}"
+    finally:
+        LOAD_GUARD.finish()
+
+
+@app.get("/api/load/limits")
+def load_limits():
+    """上限与必填项（前端展示与预校验）。"""
+    return {"requiredFields": list(load_controls.REQUIRED_PLAN_FIELDS),
+            "defaultLimits": load_controls.DEFAULT_LIMITS,
+            "hardLimits": load_controls.HARD_LIMITS,
+            "sizes": mock_db.LOAD_SIZES,
+            "guard": LOAD_GUARD.snapshot()}
+
+
+@app.post("/api/load/plan")
+def load_plan(payload: LoadPlanPayload):
+    """提交压力计划：§16 十项必填 + 上限校验 + 负责人确认（默认不改变运行状态）。"""
+    if LOAD_GUARD.state in ("running", "stopping"):
+        raise HTTPException(status_code=409,
+                            detail=f"当前状态 {LOAD_GUARD.state}：先停止本轮压测再提交新计划")
+    result = LOAD_GUARD.submit_plan(payload.model_dump())
+    if not result["ok"]:
+        raise HTTPException(status_code=400, detail={"errors": result["errors"]})
+    return result
+
+
+@app.post("/api/load/start")
+def load_start(payload: dict):
+    """显式开启压测（§16：必须由测试人员显式开启）。按通道启动对应执行器。"""
+    if LOAD_GUARD.state != "ready":
+        raise HTTPException(status_code=409,
+                            detail=f"当前状态 {LOAD_GUARD.state}：须先提交并通过压力计划校验")
+    plan = LOAD_GUARD.plan or {}
+    channel = plan["channel"]
+    run_id = LOAD_GUARD.start(plan.get("runId") or None)
+    try:
+        if channel == "API_INBOUND":
+            client = _build_inbound_client()
+            entity = (plan.get("entityCode") or payload.get("entityCode") or "").upper()
+            obj = _require_obj_by_entity(entity) or _require_obj(plan.get("objectCode") or "")
+            inbound = (obj or {}).get("apiInbound") or {}
+            if not inbound.get("supported"):
+                raise HTTPException(status_code=400, detail=f"{obj['objectCode']} 不支持 API_INBOUND")
+            entity = entity or inbound["entityCode"]
+            rows = sa.adapt_inbound_rows(obj, run_id)["rows"]
+            global LOAD_RUNNER
+            LOAD_RUNNER = runner_p3.P3Runner(
+                client, entity, rows, plan, LOAD_GUARD, LOAD_METRICS,
+                contract_version=inbound.get("contractVersion"))
+            LOAD_RUNNER.start()
+        elif channel == "DB_SYNC":
+            obj = _require_obj(plan.get("objectCode") or "")
+            db_spec = obj.get("dbSync") or {}
+            if not db_spec.get("supported") or not db_spec.get("table"):
+                raise HTTPException(status_code=400, detail=f"{obj['objectCode']} 不支持 DB_SYNC")
+            if not _env("AIDOP_SIM_MYSQL_PASSWORD"):
+                raise HTTPException(status_code=400, detail="缺少环境变量 AIDOP_SIM_MYSQL_PASSWORD")
+            thread = threading.Thread(target=_run_p1_generation, args=(obj, plan, run_id),
+                                      name=f"p1-gen-{run_id}", daemon=True)
+            DB_GEN["thread"] = thread
+            LOAD_GUARD.register_worker(thread)
+            thread.start()
+        elif channel == "API_PULL":
+            if not _mock_running():
+                raise HTTPException(status_code=400, detail="Mock API 未启动（先 POST /api/mock-api/start）")
+            _apply_p2_load(plan)
+            # P2 无后台负载：由 Ai-DOP 侧发压，压力模式由 /api/load/stop 实时停止
+        else:
+            raise HTTPException(status_code=400, detail=f"unknown channel: {channel}")
+    except HTTPException:
+        LOAD_GUARD.finish("启动失败")
+        raise
+    return {"ok": True, "runId": run_id, "channel": channel, "state": LOAD_GUARD.state}
+
+
+@app.post("/api/load/stop")
+def load_stop(payload: dict):
+    """立即停止（§21）：P1/P3 置停止信号等待在途收尾；P2 恢复 Mock 正常模式。"""
+    plan = LOAD_GUARD.plan or {}
+    reason = str(payload.get("reason") or "manual")
+    p2_recovered = False
+    if plan.get("channel") == "API_PULL":
+        _stop_p2_load()
+        p2_recovered = True
+    result = LOAD_GUARD.request_stop(reason)
+    if plan.get("channel") == "API_PULL":
+        # P2 无后台线程：直接收尾（含无残留断言）
+        result.update(LOAD_GUARD.finish(reason))
+    return {"ok": True, "p2Recovered": p2_recovered, **result}
+
+
+@app.get("/api/load/status")
+def load_status():
+    """实时状态（§21 实时基础指标 + 错误摘要）。"""
+    guard = LOAD_GUARD.snapshot()
+    out = {"guard": guard, "metrics": LOAD_METRICS.snapshot()}
+    channel = guard.get("planChannel")
+    if channel == "API_PULL":
+        out["p2"] = {"loadMode": MOCK_STATE.settings.load_mode,
+                      "stats": MOCK_STATE.load_stats.snapshot()}
+    if channel == "DB_SYNC":
+        out["p1"] = {"result": DB_GEN.get("result"), "error": DB_GEN.get("error")}
+        try:
+            out["p1"]["db"] = mock_db.db_status()
+        except Exception:  # noqa: BLE001 —— 模拟库不可达时只影响统计展示
+            out["p1"]["db"] = None
+    return out
+
+
+@app.get("/api/load/aidop-checklist/{channel}")
+def load_aidop_checklist(channel: str):
+    """Ai-DOP 人工执行和观察清单（§18 末项；§17 末句资源观察）。"""
+    common = [
+        "确认压力计划已获负责人审批，在当前时间窗口内执行",
+        "打开 Ai-DOP 监控工具，观察 CPU、内存、队列深度、数据库连接数",
+        "记录 mdp_sync_log / mdp_inbound_request 流水号与状态",
+    ]
+    if channel == "DB_SYNC":
+        steps = [
+            "在 Ai-DOP 登记 SIM_MODE3_DB 数据源并测连（连接指向模拟库）",
+            "确认实体 source_table_name 指向目标表，增量列与模拟库一致",
+            "人工触发首次全量同步，观察分批拉取与耗时",
+            "全量完成后核对 Ai-DOP 贴源表行数与模拟库源端行数（分页无丢行、无重复）",
+            "修改模拟库数据后触发增量，核对增量值与批次",
+            "同步中断后恢复，观察断点续拉状态是否可解释",
+        ]
+    elif channel == "API_PULL":
+        steps = [
+            "在 Ai-DOP 登记 SIM_MODE3_API 数据源并测连（指向 Mock API :8018）",
+            "人工触发拉取，观察多批 cursor 推进与批次耗时",
+            "核对请求量与模拟器统计（请求量/QPS/429/5xx 可对应）",
+            "观察间歇 429/5xx/延迟下 Ai-DOP 的重试与错误提示",
+            "压力结束后停止 Mock 压力模式，确认可恢复正常拉取",
+        ]
+    elif channel == "API_INBOUND":
+        steps = [
+            "观察 mdp_inbound_request 流水：受理数、拒收数与原因分类",
+            "核对贴源表 mdp_stg_* 新增行数与发送行数",
+            "观察 429 限流响应与恢复后继续受理的情况",
+            "压测结束后确认无持续入站请求（模拟器停止后 Ai-DOP 侧连接回落）",
+        ]
+    else:
+        raise HTTPException(status_code=404, detail="unknown channel")
+    return {"channel": channel, "steps": common + steps}
+
+
+LOAD_REPORT_FIELDS = [
+    ("runId", "运行编号 runId"),
+    ("channel", "集成方式"),
+    ("environment", "环境"),
+    ("tenant", "测试租户"),
+    ("owner", "负责人"),
+    ("timeWindow", "时间窗口"),
+    ("maxDataRows", "最大数据量"),
+    ("maxConcurrency", "最大并发"),
+    ("maxQps", "最大 QPS"),
+    ("maxDurationSeconds", "最大持续时间"),
+    ("stopThreshold", "停止阈值"),
+    ("cleanupPlan", "清理方案"),
+    ("startedAt", "实际开始时间"),
+    ("finishedAt", "实际结束时间"),
+    ("stopReason", "停止原因"),
+    ("aidopResource", "Ai-DOP 资源人工观察"),
+    ("aidopFlow", "Ai-DOP 流水核对"),
+    ("conclusion", "结论"),
+    ("notes", "备注"),
+]
+
+
+def build_load_report_markdown(entry: dict) -> str:
+    """压力报告导出 Markdown（§17 指标全列 + §16 计划 + 人工观察 + 脱敏声明）。"""
+    lines = ["# Ai-DOP 第三方系统模拟器 · 压力测试报告", ""]
+    metrics = entry.get("metrics") or {}
+    guard = entry.get("guard") or {}
+    lines.append(f"- 集成方式：{CHANNEL_LABELS.get(entry.get('channel', ''), entry.get('channel', ''))}")
+    lines.append(f"- runId：{entry.get('runId', '')}")
+    lines.append("")
+    lines.append("## 测试指标（§17）")
+    lines.append("| 指标 | 值 |")
+    lines.append("|---|---|")
+    for key, label in load_metrics.METRIC_FIELDS:
+        value = metrics.get(key)
+        if isinstance(value, dict):
+            value = ", ".join(f"{k}:{v}" for k, v in value.items()) or "—"
+        lines.append(f"| {label} | {value if value not in (None, '') else '—'} |")
+    lines.append("")
+    lines.append("## 计划与确认（§16）")
+    lines.append("| 项目 | 内容 |")
+    lines.append("|---|---|")
+    for key, label in LOAD_REPORT_FIELDS:
+        value = entry.get(key)
+        if key in ("startedAt", "finishedAt", "stopReason"):
+            value = guard.get(key, value)
+        value = str(value or "").strip() or "（未填写）"
+        lines.append(f"| {label} | {value.replace(chr(10), '<br>')} |")
+    lines.append("")
+    lines.append(f"报告 ID：{entry.get('id', '')} · 创建时间：{entry.get('createdAt', '')}")
+    lines.append("")
+    lines.append("> 脱敏说明：本报告已自动脱敏，AccessKey 仅保留后四位；"
+                 "数据库密码、Secret、JWT、完整 Authorization 与连接串已遮蔽。")
+    return "\n".join(lines)
+
+
+@app.post("/api/load/report")
+def load_report_save(payload: LoadReportPayload):
+    """保存压力报告：自动快照计划与指标 + 人工观察字段，保存前脱敏。"""
+    guard = LOAD_GUARD.snapshot()
+    fields = dict(payload.model_dump())
+    fields.update({
+        "metrics": LOAD_METRICS.snapshot(),
+        "guard": guard,
+        "environment": (LOAD_GUARD.plan or {}).get("environment", ""),
+        "tenant": (LOAD_GUARD.plan or {}).get("tenant", ""),
+        "owner": (LOAD_GUARD.plan or {}).get("owner", ""),
+        "timeWindow": (LOAD_GUARD.plan or {}).get("timeWindow", ""),
+        "maxDataRows": (LOAD_GUARD.plan or {}).get("maxDataRows", ""),
+        "maxConcurrency": (LOAD_GUARD.plan or {}).get("maxConcurrency", ""),
+        "maxQps": (LOAD_GUARD.plan or {}).get("maxQps", ""),
+        "maxDurationSeconds": (LOAD_GUARD.plan or {}).get("maxDurationSeconds", ""),
+        "stopThreshold": (LOAD_GUARD.plan or {}).get("stopThreshold", ""),
+        "cleanupPlan": (LOAD_GUARD.plan or {}).get("cleanupPlan", ""),
+    })
+    entry = LOAD_REPORTS.save(fields, _secrets(), _access_keys())
+    return {"ok": True, "report": entry}
+
+
+@app.get("/api/load/report/list")
+def load_report_list():
+    return {"reports": LOAD_REPORTS.list()}
+
+
+@app.get("/api/load/report/export/{report_id}")
+def load_report_export(report_id: str):
+    entry = LOAD_REPORTS.get(report_id)
+    if not entry:
+        raise HTTPException(status_code=404, detail="report not found")
+    md = build_load_report_markdown(entry)
+    filename = f"aidop-sim-load-report-{report_id}.md"
+    return Response(content=md, media_type="text/markdown; charset=utf-8",
+                    headers={"Content-Disposition": f'attachment; filename="{filename}"'})
+
+
+@app.delete("/api/load/report/{report_id}")
+def load_report_delete(report_id: str):
+    if not LOAD_REPORTS.delete(report_id):
+        raise HTTPException(status_code=404, detail="report not found")
+    return {"ok": True}
+
+
 # ---------------------------------------------------------------------------
 # 全链路报告（F5）
 # ---------------------------------------------------------------------------

tools/integration-simulator/load/__init__.py

@@ -0,0 +1,6 @@
+"""压力测试能力包（任务书第三阶段 §16-§21）。
+
+- controls：安全控制（默认关闭、十项确认、上限、自动停止、立即停止、无残留）；
+- metrics：指标采集与计算（§17 全量指标，只采集第三方侧可观察项）；
+- runner_p3：方式丙压测执行器（模拟器是负载发起方）。
+"""

tools/integration-simulator/load/controls.py

@@ -0,0 +1,263 @@
+"""压力测试安全控制（任务书 §16 / §20 / §21 验收表）。
+
+不变量（对应第三阶段验收表）：
+- 压力测试默认关闭：LoadGuard 初始 idle，无完整计划不可启动；
+- §16 执行前十项必填，缺一拒绝；共享环境须负责人确认（confirmed）；
+- 参数保守上限，提高须 escalationConfirm 显式二次确认，且不得越过硬上限；
+- 达到连续错误 / 超时 / 429 阈值自动停止；不自动无限重试；
+- 立即停止：stop 信号 + 在途请求收尾等待 + 无残留负载断言；
+- 每轮唯一 runId；P3 行构造禁止携带 tenantId（§20 安全限制）。
+"""
+from __future__ import annotations
+
+import threading
+import time
+import uuid
+
+CHANNELS = ("DB_SYNC", "API_PULL", "API_INBOUND")
+
+# §16 执行前必填项（十项）
+REQUIRED_PLAN_FIELDS = (
+    "environment",          # 环境
+    "tenant",               # 测试租户
+    "owner",                # 负责人
+    "timeWindow",           # 时间窗口
+    "maxDataRows",          # 最大数据量
+    "maxConcurrency",       # 最大并发
+    "maxQps",               # 最大 QPS
+    "maxDurationSeconds",   # 最大持续时间
+    "stopThreshold",        # 停止阈值（如：连续 5 次错误/超时/429 自动停止）
+    "cleanupPlan",          # 清理方案
+)
+
+# 保守默认上限（§20 安全限制）。提高须 escalationConfirm=True，且不得超过 HARD_LIMITS
+DEFAULT_LIMITS = {
+    "maxDataRows": 100_000,        # 与 L 档一致；XL 须审批
+    "maxConcurrency": 20,
+    "maxQps": 50,
+    "maxDurationSeconds": 600,
+    "maxBatchSize": 100,
+    "errorRate": 0.5,
+    "maxRetries": 3,
+    "timeoutSeconds": 30,
+    "maxConsecutiveErrors": 5,
+    "maxTimeouts": 5,
+    "maxRateLimited": 5,
+    "rateLimit": 200,          # P2 Mock 每秒最大请求数
+    "randomDelay": 10.0,       # P2 随机延迟上限（秒）
+}
+
+# 硬上限：即使显式二次确认也不允许越过（非测试环境阻断的兜底）
+HARD_LIMITS = {
+    "maxDataRows": 200_000,
+    "maxConcurrency": 50,
+    "maxQps": 100,
+    "maxDurationSeconds": 1800,
+    "maxBatchSize": 500,
+    "errorRate": 0.9,
+    "maxRetries": 5,
+    "timeoutSeconds": 120,
+    "maxConsecutiveErrors": 100,
+    "maxTimeouts": 100,
+    "maxRateLimited": 100,
+    "rateLimit": 1000,
+    "randomDelay": 60.0,
+}
+
+# 行内禁止携带的租户字段（§20：请求体不得携带任意 tenantId 改变目标租户）
+_TENANT_FIELD_NAMES = ("tenantid", "tenant_id", "tenantcode", "tenant_code",
+                       "domainid", "domain_id", "orgid", "org_id")
+
+
+def strip_tenant_fields(row: dict) -> dict:
+    """剔除行内租户/组织字段，防止 payload 切换目标租户。"""
+    return {k: v for k, v in row.items() if str(k).lower() not in _TENANT_FIELD_NAMES}
+
+
+class ValidationError(ValueError):
+    pass
+
+
+def validate_plan(plan: dict) -> list[str]:
+    """校验压力计划，返回错误列表（空列表 = 通过）。"""
+    errors: list[str] = []
+    if plan.get("channel") not in CHANNELS:
+        errors.append("channel 必须是 DB_SYNC / API_PULL / API_INBOUND")
+    for field in REQUIRED_PLAN_FIELDS:
+        if plan.get(field) in (None, "", 0):
+            errors.append(f"缺少必填项：{field}（§16 执行前必须填写）")
+    if not bool(plan.get("confirmed")):
+        errors.append("共享开发/UAT 环境须由负责人确认后执行（confirmed=true）")
+    escalated = bool(plan.get("escalationConfirm"))
+    float_keys = ("errorRate", "randomDelay")
+    for key, cap in DEFAULT_LIMITS.items():
+        raw = plan.get(key)
+        if raw in (None, "", 0):
+            continue
+        try:
+            value = float(raw) if key in float_keys else int(raw)
+        except (TypeError, ValueError):
+            errors.append(f"{key} 必须是数字，收到：{raw!r}")
+            continue
+        if key == "errorRate" and not 0 <= value <= 1:
+            errors.append(f"errorRate 必须在 0-1 之间")
+            continue
+        if value > cap and not escalated:
+            errors.append(f"{key}={value} 超过保守上限 {cap}，需勾选「提高上限二次确认」")
+        if value > HARD_LIMITS[key]:
+            errors.append(f"{key}={value} 超过硬上限 {HARD_LIMITS[key]}，即使二次确认也不允许")
+    return errors
+
+
+class AutoStopTracker:
+    """自动停止判定（§20 安全限制：连续错误、超时或 429 达阈值自动停止）。"""
+
+    def __init__(self, consecutive_errors: int = 5, timeouts: int = 5, rate_limited: int = 5):
+        self.max_consecutive_errors = max(1, consecutive_errors)
+        self.max_timeouts = max(1, timeouts)
+        self.max_rate_limited = max(1, rate_limited)
+        self._consecutive_errors = 0
+        self._timeouts = 0
+        self._rate_limited = 0
+        self._tripped_by: str | None = None
+        self._lock = threading.Lock()
+
+    def record(self, status: int) -> bool:
+        """记录一次请求结果；返回 True 表示应自动停止。
+
+        status 约定：0 = 网络超时/不可达；429 = 限流；其余为 HTTP 状态码。
+        """
+        with self._lock:
+            if status == 0:
+                self._timeouts += 1
+                self._consecutive_errors += 1
+            elif status == 429:
+                self._rate_limited += 1
+                self._consecutive_errors += 1
+            elif status >= 400:
+                self._consecutive_errors += 1
+            else:
+                self._consecutive_errors = 0
+            if self._tripped_by is None:
+                if self._consecutive_errors >= self.max_consecutive_errors:
+                    self._tripped_by = "consecutiveErrors"
+                elif self._timeouts >= self.max_timeouts:
+                    self._tripped_by = "timeouts"
+                elif self._rate_limited >= self.max_rate_limited:
+                    self._tripped_by = "rateLimited"
+            return self._tripped_by is not None
+
+    @property
+    def tripped_by(self) -> str | None:
+        return self._tripped_by
+
+    def summary(self) -> dict:
+        return {"consecutiveErrors": self._consecutive_errors,
+                "timeouts": self._timeouts, "rateLimited": self._rate_limited,
+                "thresholds": {"maxConsecutiveErrors": self.max_consecutive_errors,
+                               "maxTimeouts": self.max_timeouts,
+                               "maxRateLimited": self.max_rate_limited},
+                "trippedBy": self._tripped_by}
+
+
+class LoadGuard:
+    """压力测试生命周期守卫（进程内单例，重启即默认关闭）。
+
+    状态机：idle -> ready -> running -> stopping -> stopped
+    """
+
+    def __init__(self):
+        self.state = "idle"            # idle | ready | running | stopping | stopped
+        self.plan: dict | None = None
+        self.run_id: str | None = None
+        self.started_at: str | None = None
+        self.finished_at: str | None = None
+        self.stop_reason: str | None = None
+        self.stop = threading.Event()
+        self._workers: list[threading.Thread] = []
+        self._lock = threading.Lock()
+
+    # -- 计划 ----------------------------------------------------------------
+
+    def submit_plan(self, plan: dict) -> dict:
+        """保存并校验计划。校验失败不改变当前状态。"""
+        errors = validate_plan(plan)
+        if errors:
+            return {"ok": False, "errors": errors, "state": self.state}
+        with self._lock:
+            self.plan = dict(plan)
+            self.state = "ready"
+            self.stop.clear()
+            self.stop_reason = None
+            self.finished_at = None
+        return {"ok": True, "state": self.state, "runId": self.plan.get("runId") or ""}
+
+    # -- 生命周期 ------------------------------------------------------------
+
+    def start(self, run_id: str | None = None) -> str:
+        """ready -> running；返回本轮 runId（每轮唯一）。"""
+        with self._lock:
+            if self.state != "ready":
+                raise ValidationError(f"当前状态 {self.state}，无法开始（须先提交并通过计划校验）")
+            self.run_id = run_id or f"LOAD-{uuid.uuid4().hex[:12]}"
+            self.started_at = time.strftime("%Y-%m-%dT%H:%M:%S")
+            self.state = "running"
+            self.stop.clear()
+            return self.run_id
+
+    def request_stop(self, reason: str = "manual") -> dict:
+        """请求停止：置信号，状态 -> stopping（在途请求由执行器收尾）。"""
+        with self._lock:
+            self.stop_reason = self.stop_reason or reason
+            if self.state in ("ready", "running"):
+                self.state = "stopping"
+        self.stop.set()
+        return {"ok": True, "state": self.state, "reason": self.stop_reason}
+
+    def finish(self, reason: str | None = None) -> dict:
+        """执行器收尾调用：stopping/running -> stopped，并断言无残留负载。"""
+        with self._lock:
+            if self.state not in ("running", "stopping"):
+                raise ValidationError(f"当前状态 {self.state}，无法收尾")
+            # 排除当前线程：P1/P3 执行器线程会注册自身为 worker，
+            # 它在 finish() 调用瞬间必然存活，不应误报为残留
+            current = threading.current_thread()
+            residual = [t.name for t in self._workers
+                        if t.is_alive() and t is not current]
+            if residual:
+                # 在途收尾不彻底：如实上报，不伪装干净
+                self.state = "stopped"
+                self.finished_at = time.strftime("%Y-%m-%dT%H:%M:%S")
+                self.stop_reason = self.stop_reason or reason or "finished"
+                return {"ok": False, "state": self.state,
+                        "reason": self.stop_reason or "finished",
+                        "residualThreads": residual}
+            self.state = "stopped"
+            self.finished_at = time.strftime("%Y-%m-%dT%H:%M:%S")
+            self.stop_reason = self.stop_reason or reason or "finished"
+            return {"ok": True, "state": self.state, "reason": self.stop_reason,
+                    "residualThreads": []}
+
+    def register_worker(self, worker: threading.Thread) -> None:
+        with self._lock:
+            self._workers.append(worker)
+
+    def no_residual(self) -> bool:
+        with self._lock:
+            return all(not t.is_alive() for t in self._workers)
+
+    # -- 快照 ----------------------------------------------------------------
+
+    def snapshot(self) -> dict:
+        with self._lock:
+            return {
+                "state": self.state,
+                "runId": self.run_id,
+                "startedAt": self.started_at,
+                "finishedAt": self.finished_at,
+                "stopReason": self.stop_reason,
+                "planChannel": (self.plan or {}).get("channel"),
+                "planConfirmed": bool((self.plan or {}).get("confirmed")),
+                # 持锁内直接检查，避免重入 no_residual() 死锁（Lock 非重入）
+                "noResidual": all(not t.is_alive() for t in self._workers),
+            }

tools/integration-simulator/load/metrics.py

@@ -0,0 +1,197 @@
+"""压力指标采集与计算（任务书 §17）。
+
+只采集第三方侧可观察指标：请求量、批次、行数、成功率、QPS、行吞吐、
+延迟分布（P50/P95/P99）、HTTP 错误分类、超时数、限流数、首次失败时间、起止时间。
+Ai-DOP CPU/内存/队列/数据库资源由测试人员在现有监控工具观察（§17 末句）。
+"""
+from __future__ import annotations
+
+import threading
+import time
+
+# §17 指标快照字段（导出报告按此列全量输出）
+METRIC_FIELDS = (
+    ("totalRequests", "总请求数"),
+    ("totalBatches", "总批次数"),
+    ("totalRows", "总数据行数"),
+    ("success", "成功数"),
+    ("failed", "失败数"),
+    ("successRate", "成功率"),
+    ("actualQps", "实际 QPS"),
+    ("rowThroughput", "行吞吐（行/秒）"),
+    ("avgMs", "平均耗时（ms）"),
+    ("p50Ms", "P50（ms）"),
+    ("p95Ms", "P95（ms）"),
+    ("p99Ms", "P99（ms）"),
+    ("maxMs", "最大耗时（ms）"),
+    ("httpErrors", "HTTP 错误分类"),
+    ("timeouts", "超时数"),
+    ("rateLimited", "限流数"),
+    ("firstFailureAt", "首次失败时间"),
+    ("startedAt", "开始时间"),
+    ("finishedAt", "结束时间"),
+)
+
+
+def percentile(sorted_values: list[float], p: float) -> float:
+    """线性插值百分位（P50/P95/P99）。空列表返回 0.0。"""
+    if not sorted_values:
+        return 0.0
+    if p <= 0:
+        return sorted_values[0]
+    if p >= 100:
+        return sorted_values[-1]
+    k = (len(sorted_values) - 1) * p / 100.0
+    lo = int(k)
+    hi = min(lo + 1, len(sorted_values) - 1)
+    return sorted_values[lo] + (sorted_values[hi] - sorted_values[lo]) * (k - lo)
+
+
+class LoadMetrics:
+    """线程安全的压力指标累计器。
+
+    QPS 口径：最近 window_seconds 内完成的请求数 / 窗口秒数（滚动窗口，
+    长任务结束后与实际总吞吐可解释比对）。
+    """
+
+    def __init__(self, window_seconds: float = 60.0):
+        self.window_seconds = max(1.0, window_seconds)
+        self.started_at: str | None = None
+        self.finished_at: str | None = None
+        self.total_requests = 0
+        self.total_batches = 0
+        self.total_rows = 0
+        self.success = 0
+        self.failed = 0
+        self.timeouts = 0
+        self.rate_limited = 0
+        self.http_errors: dict[int, int] = {}
+        self.latencies: list[float] = []
+        self.first_failure_at: str | None = None
+        self._completion_times: list[float] = []
+        self._final_qps: float | None = None
+        self._final_throughput: float | None = None
+        self._lock = threading.Lock()
+
+    def reset(self) -> None:
+        """清空全部累计（新一轮压测开始前由 mark_start 调用）。"""
+        self.started_at = None
+        self.finished_at = None
+        self.total_requests = 0
+        self.total_batches = 0
+        self.total_rows = 0
+        self.success = 0
+        self.failed = 0
+        self.timeouts = 0
+        self.rate_limited = 0
+        self.http_errors = {}
+        self.latencies = []
+        self.first_failure_at = None
+        self._completion_times = []
+        self._final_qps = None
+        self._final_throughput = None
+
+    def mark_start(self) -> None:
+        with self._lock:
+            # 单例跨轮复用：新一轮压测开始即重置上一轮累计
+            self.reset()
+            self.started_at = time.strftime("%Y-%m-%dT%H:%M:%S")
+
+    def mark_finish(self) -> None:
+        with self._lock:
+            if self.finished_at is None:
+                self.finished_at = time.strftime("%Y-%m-%dT%H:%M:%S")
+                # 结束后冻结滚动指标：QPS/行吞吐定格在结束时刻，不再随时间衰减
+                now = time.monotonic()
+                self._final_qps = self._actual_qps_locked(now)
+                self._final_throughput = self._row_throughput_locked()
+
+    def record(self, status: int, latency_s: float, rows: int = 0, batches: int = 1) -> None:
+        """记录一次请求/批次结果。status：0=超时/不可达，429=限流，其余 HTTP。"""
+        now = time.monotonic()
+        with self._lock:
+            self.total_requests += 1
+            self.total_batches += batches
+            self.total_rows += max(0, rows)
+            self.latencies.append(max(0.0, latency_s))
+            self._completion_times.append(now)
+            if status == 0:
+                self.timeouts += 1
+                self.failed += 1
+            elif status == 429:
+                self.rate_limited += 1
+                self.failed += 1
+            elif 200 <= status < 300 or status == 202:
+                self.success += 1
+            else:
+                self.failed += 1
+                self.http_errors[status] = self.http_errors.get(status, 0) + 1
+            if (status == 0 or status >= 400) and self.first_failure_at is None:
+                self.first_failure_at = time.strftime("%Y-%m-%dT%H:%M:%S")
+
+    # -- 计算 ----------------------------------------------------------------
+
+    def _pruned_completion_times(self, now: float) -> list[float]:
+        cutoff = now - self.window_seconds
+        return [t for t in self._completion_times if t >= cutoff]
+
+    def _actual_qps_locked(self, now: float) -> float:
+        recent = self._pruned_completion_times(now)
+        if not recent:
+            return 0.0
+        span = now - recent[0]
+        if span <= 0:
+            return 0.0
+        return round(len(recent) / span, 2)
+
+    def actual_qps(self) -> float:
+        now = time.monotonic()
+        with self._lock:
+            return self._actual_qps_locked(now)
+
+    def _row_throughput_locked(self) -> float:
+        if not self.started_at or self.total_rows == 0:
+            return 0.0
+        end = time.strptime(self.finished_at, "%Y-%m-%dT%H:%M:%S") if self.finished_at else time.localtime()
+        start = time.strptime(self.started_at, "%Y-%m-%dT%H:%M:%S")
+        elapsed = max(1e-9, time.mktime(end) - time.mktime(start))
+        return round(self.total_rows / elapsed, 2)
+
+    def row_throughput(self) -> float:
+        with self._lock:
+            return self._row_throughput_locked()
+
+    # -- 快照 ----------------------------------------------------------------
+
+    def snapshot(self) -> dict:
+        with self._lock:
+            latencies = sorted(self.latencies)
+            total = self.total_requests
+            success_rate = round(self.success / total, 4) if total else 0.0
+            # 持锁内用 *_locked 计算，避免重入锁死（Lock 非重入）
+            now = time.monotonic()
+            frozen = self.finished_at is not None
+            return {
+                "totalRequests": total,
+                "totalBatches": self.total_batches,
+                "totalRows": self.total_rows,
+                "success": self.success,
+                "failed": self.failed,
+                "successRate": success_rate,
+                "actualQps": (self._final_qps if frozen and self._final_qps is not None
+                              else self._actual_qps_locked(now)),
+                "rowThroughput": (self._final_throughput
+                                  if frozen and self._final_throughput is not None
+                                  else self._row_throughput_locked()),
+                "avgMs": round(sum(latencies) / len(latencies) * 1000, 1) if latencies else 0.0,
+                "p50Ms": round(percentile(latencies, 50) * 1000, 1),
+                "p95Ms": round(percentile(latencies, 95) * 1000, 1),
+                "p99Ms": round(percentile(latencies, 99) * 1000, 1),
+                "maxMs": round(max(latencies) * 1000, 1) if latencies else 0.0,
+                "httpErrors": dict(sorted(self.http_errors.items())),
+                "timeouts": self.timeouts,
+                "rateLimited": self.rate_limited,
+                "firstFailureAt": self.first_failure_at,
+                "startedAt": self.started_at,
+                "finishedAt": self.finished_at,
+            }

tools/integration-simulator/load/runner_p3.py

@@ -0,0 +1,182 @@
+"""方式丙压测执行器（任务书 §20）：模拟器是负载发起方。
+
+能力映射：
+- 并发 worker（ThreadPoolExecutor，plan.maxConcurrency）；
+- QPS 限制（槽位法节流器，实际 QPS 接近目标且偏差可解释）；
+- 批大小（bulk 模式每请求行数 = plan.batchSize）；
+- 持续时间 / 阶梯配置（rampSteps，缺省固定档）；
+- 错误数据比例（errorRate 概率缺必填字段，与 N3 partial-reject 同语义）；
+- 幂等重复（idemReplay：复用同一 idempotencyKey 重发）；
+- 立即停止（guard.stop 信号，不再调度新请求，等待在途收尾）；
+- 达到连续错误/超时/429 阈值自动停止（AutoStopTracker）；
+- 每轮唯一 runId；行构造剔除租户字段（§20 安全限制）。
+"""
+from __future__ import annotations
+
+import random
+import sys
+import threading
+import time
+from concurrent.futures import ThreadPoolExecutor
+from pathlib import Path
+
+_SIM_ROOT = Path(__file__).resolve().parents[1]
+if str(_SIM_ROOT) not in sys.path:
+    sys.path.insert(0, str(_SIM_ROOT))
+
+from catalog import sample_adapters as sa  # noqa: E402
+from load import controls  # noqa: E402
+from load.metrics import LoadMetrics  # noqa: E402
+
+
+class _RateLimiter:
+    """固定 QPS 节流器（槽位法：请求间隔 = 1/QPS）。
+
+    近似实现，实际 QPS 受网络与并发影响，偏差在报告中可解释（§20 验收）。
+    """
+
+    def __init__(self, qps: float):
+        self.interval = 1.0 / qps if qps > 0 else 0.0
+        self._next = 0.0
+        self._lock = threading.Lock()
+
+    def wait(self) -> None:
+        if self.interval <= 0:
+            return
+        with self._lock:
+            now = time.monotonic()
+            slot = self._next if now < self._next else now
+            self._next = slot + self.interval
+        delay = slot - now
+        if delay > 0:
+            time.sleep(delay)
+
+
+class P3Runner(threading.Thread):
+    """后台线程运行方式丙压测；结束后把 LoadGuard 收尾（含无残留断言）。"""
+
+    def __init__(self, client, entity: str, template_rows: list[dict], plan: dict,
+                 guard: controls.LoadGuard, metrics: LoadMetrics, *,
+                 contract_version: str | None = None):
+        super().__init__(name=f"p3-runner-{plan.get('runId', 'x')}", daemon=True)
+        self.client = client
+        self.entity = entity
+        self.template_rows = [dict(r) for r in template_rows]
+        self.plan = plan
+        self.guard = guard
+        self.metrics = metrics
+        self.contract_version = contract_version
+        self.tracker = controls.AutoStopTracker(
+            int(plan.get("maxConsecutiveErrors") or 5),
+            int(plan.get("maxTimeouts") or 5),
+            int(plan.get("maxRateLimited") or 5))
+        self.mode = plan.get("mode") or "push"            # push | bulk
+        self.batch_size = max(1, int(plan.get("batchSize") or 1))
+        self.error_rate = float(plan.get("errorRate") or 0.0)
+        self.idem_replay = bool(plan.get("idemReplay"))
+        self._first_required = self._resolve_first_required_field()
+        self._replay_idem = f"LOAD-REPLAY-{random.randrange(1 << 32):08x}"
+
+    # -- 行构造 --------------------------------------------------------------
+
+    def _resolve_first_required_field(self) -> str | None:
+        spec = sa.INBOUND_SPECS.get(self.entity) or []
+        for field, required, _ in spec:
+            if required:
+                return field
+        return None
+
+    def _build_row(self, seq: int) -> dict:
+        """固定/随机样例 + 序号唯一化 + 租户字段剔除。"""
+        row = dict(random.choice(self.template_rows)) if self.plan.get("sampleMode") == "random" \
+            else dict(self.template_rows[seq % len(self.template_rows)])
+        row = controls.strip_tenant_fields(row)
+        for key in ("bizKey", "Id"):
+            if key in row:
+                row[key] = f"{row[key]}#{self.guard.run_id}-{seq}"
+        row["sourceUpdatedAt"] = sa.now_iso()
+        return row
+
+    def _one_request(self, seq: int) -> None:
+        rows = [self._build_row(seq)]
+        bad = self.error_rate > 0 and random.random() < self.error_rate
+        if bad and self._first_required:
+            # 部分错误数据：缺必填字段（同 N3 partial-reject 语义）
+            rows[0].pop(self._first_required, None)
+        started = time.monotonic()
+        kwargs: dict = {}
+        if self.idem_replay:
+            kwargs["idem"] = self._replay_idem
+        try:
+            if self.mode == "bulk":
+                if self.batch_size > 1:
+                    rows.extend(self._build_row(seq * 1000 + i) for i in range(1, self.batch_size))
+                lines = [{"idempotencyKey": f"{self.guard.run_id}-L{seq}-{i}",
+                          "data": {"list": [r]}} for i, r in enumerate(rows)]
+                status, _, _ = self.client.op_bulk(self.entity, lines,
+                                                   contract_version=self.contract_version)
+            else:
+                envelope = sa.inbound_envelope(rows)
+                status, _, _ = self.client.op_push(self.entity, envelope,
+                                                   contract_version=self.contract_version, **kwargs)
+            latency = time.monotonic() - started
+            self.metrics.record(status, latency, rows=len(rows))
+            if self.tracker.record(status):
+                self.guard.request_stop(f"自动停止：{self.tracker.tripped_by} 达到阈值")
+        except Exception as ex:  # noqa: BLE001 —— 压测收集异常而非中断
+            self.metrics.record(0, time.monotonic() - started, rows=len(rows))
+            self.guard.stop_reason = self.guard.stop_reason or f"worker 异常：{type(ex).__name__}"
+            if self.tracker.record(0):
+                self.guard.request_stop(f"自动停止：{self.tracker.tripped_by} 达到阈值")
+
+    # -- 阶梯 ----------------------------------------------------------------
+
+    def _ramp_steps(self) -> list[dict]:
+        steps = self.plan.get("rampSteps") or []
+        if steps:
+            return steps
+        return [{
+            "duration": float(self.plan.get("maxDurationSeconds") or 10),
+            "qps": float(self.plan.get("maxQps") or 1),
+            "concurrency": int(self.plan.get("maxConcurrency") or 1),
+        }]
+
+    def _run_step(self, step: dict) -> None:
+        duration = max(0.5, float(step.get("duration") or 1))
+        qps = max(0.0, float(step.get("qps") or 0))
+        concurrency = max(1, int(step.get("concurrency") or 1))
+        limiter = _RateLimiter(qps)
+        deadline = time.monotonic() + duration
+        futures = []
+        seq = 0
+        with ThreadPoolExecutor(max_workers=concurrency) as pool:
+            while time.monotonic() < deadline and not self.guard.stop.is_set():
+                limiter.wait()
+                if self.guard.stop.is_set():
+                    break
+                seq += 1
+                futures.append(pool.submit(self._one_request, seq))
+                if len(futures) >= concurrency * 8:
+                    futures = [f for f in futures if not f.done()]
+            # 停止后：取消排队未启动的任务，只等待真正在途的请求收尾（§20）
+            pool.shutdown(wait=True, cancel_futures=True)
+
+    def run(self) -> None:
+        self.guard.register_worker(self)
+        warmup = max(0, int(self.plan.get("warmupSeconds") or 0))
+        if warmup and not self.guard.stop.is_set():
+            # §17 预热时间：等待环境稳定后再发压（可被立即停止打断）
+            self.guard.stop.wait(timeout=warmup)
+        self.metrics.mark_start()
+        try:
+            for step in self._ramp_steps():
+                if self.guard.stop.is_set():
+                    break
+                self._run_step(step)
+        finally:
+            self.metrics.mark_finish()
+            result = self.guard.finish()
+            if not result.get("ok"):
+                # 在途线程未收尾：记录到 stop_reason 供报告如实呈现
+                self.guard.stop_reason = (self.guard.stop_reason or "") + \
+                    f"（残留线程：{result.get('residualThreads')}）"

tools/integration-simulator/providers/mock_api.py

@@ -14,8 +14,11 @@ from __future__ import annotations
 
 import base64
 import json
+import random
 import sys
+import threading
 import time
+from collections import deque
 from dataclasses import dataclass, field
 from pathlib import Path
 
@@ -48,6 +51,13 @@ class MockApiSettings:
     content_type: str | None = None   # 覆盖响应 Content-Type（如 text/html）
     ignore_cursor: bool = False       # 无视 cursor 返回全部行（cursor 不推进）
     no_sort: bool = False             # 不按 bizKey 排序（cursor 乱序）
+    # ---- 第三阶段 P2 压力模式（受控：须压力计划确认后由控制台注入）----
+    load_mode: bool = False           # 压力模式开关（§19 实时停止/恢复正常）
+    page_size: int = 0                # >0: 每次仅返回前 N 行（配置每次返回行数）
+    total_rows_limit: int = 0         # >0: 响应数据总量上限（模拟大表截断）
+    random_delay: float = 0.0         # >0: 每请求随机延迟 0..random_delay 秒
+    error_rate: float = 0.0           # 0..1: 随机 500 比例（间歇 5xx）
+    rate_limit: float = 0.0           # >0: 每秒最大请求数，超出返回 429（限流阈值）
 
     def reset_faults(self) -> None:
         """恢复默认：清空全部异常注入设置（鉴权/凭据类保留）。"""
@@ -61,6 +71,66 @@ class MockApiSettings:
         self.content_type = None
         self.ignore_cursor = False
         self.no_sort = False
+        self.load_mode = False
+        self.page_size = 0
+        self.total_rows_limit = 0
+        self.random_delay = 0.0
+        self.error_rate = 0.0
+        self.rate_limit = 0.0
+
+
+@dataclass
+class MockApiLoadStats:
+    """P2 压力模式请求统计（§19：统计收到的请求量、并发、QPS、耗时和错误）。"""
+    requests: int = 0
+    active: int = 0                 # 当前并发
+    peak_active: int = 0            # 峰值并发
+    status_counts: dict = field(default_factory=dict)
+    latencies: list = field(default_factory=list)
+    rate_limited: int = 0           # 429 次数
+    recent_times: deque = field(default_factory=lambda: deque(maxlen=500))
+    _lock: threading.Lock = field(default_factory=threading.Lock)
+
+    def begin(self):
+        with self._lock:
+            self.active += 1
+            self.peak_active = max(self.peak_active, self.active)
+
+    def end(self, status: int, elapsed_s: float):
+        with self._lock:
+            self.active = max(0, self.active - 1)
+            self.requests += 1
+            self.status_counts[status] = self.status_counts.get(status, 0) + 1
+            self.latencies.append(elapsed_s)
+            self.recent_times.append(time.monotonic())
+            if status == 429:
+                self.rate_limited += 1
+
+    def snapshot(self) -> dict:
+        with self._lock:
+            latencies = sorted(self.latencies) if self.latencies else []
+            times = list(self.recent_times)
+            qps = 0.0
+            if len(times) >= 2:
+                span = times[-1] - times[0]
+                qps = round(len(times) / span, 2) if span > 0 else 0.0
+            avg = round(sum(latencies) / len(latencies) * 1000, 1) if latencies else 0.0
+            return {"requests": self.requests, "active": self.active,
+                    "peakConcurrency": self.peak_active,
+                    "qpsWindow": qps, "avgMs": avg,
+                    "maxMs": round(max(latencies) * 1000, 1) if latencies else 0.0,
+                    "statusCounts": dict(sorted(self.status_counts.items())),
+                    "rateLimited": self.rate_limited}
+
+    def reset(self):
+        with self._lock:
+            self.requests = 0
+            self.active = 0
+            self.peak_active = 0
+            self.status_counts.clear()
+            self.latencies.clear()
+            self.rate_limited = 0
+            self.recent_times.clear()
 
 
 @dataclass
@@ -68,6 +138,8 @@ class MockApiState:
     settings: MockApiSettings = field(default_factory=MockApiSettings)
     overrides: dict[str, list[dict]] = field(default_factory=dict)
     audit: list[dict] = field(default_factory=list)
+    load_stats: MockApiLoadStats = field(default_factory=MockApiLoadStats)
+    rate_hits: deque = field(default_factory=deque)   # P2 限流滑动窗口
 
     def secrets(self) -> list[str]:
         s = self.settings
@@ -120,6 +192,17 @@ def _check_auth(state: MockApiState, request: Request) -> None:
         raise HTTPException(status_code=401, detail=f"unsupported auth mode on mock side: {mode}")
 
 
+def _rate_allowed(state: MockApiState, rps: float) -> bool:
+    """滑动窗口限流：每秒最多 rps 个请求，超出返回 False（间歇 429，§19）。"""
+    now = time.monotonic()
+    while state.rate_hits and state.rate_hits[0] <= now - 1.0:
+        state.rate_hits.popleft()
+    if len(state.rate_hits) >= rps:
+        return False
+    state.rate_hits.append(now)
+    return True
+
+
 def _active_faults(s: MockApiSettings) -> list[str]:
     """当前生效的异常注入清单（写入审计，便于人工核对「选择的异常」）。"""
     faults: list[str] = []
@@ -179,6 +262,9 @@ def create_app(state: MockApiState | None = None) -> FastAPI:
             "timeoutSeconds": s.timeout_seconds, "emptyResponse": s.empty_response,
             "invalidJson": s.invalid_json, "contentType": s.content_type,
             "ignoreCursor": s.ignore_cursor, "noSort": s.no_sort,
+            "loadMode": s.load_mode, "pageSize": s.page_size,
+            "totalRowsLimit": s.total_rows_limit, "randomDelay": s.random_delay,
+            "errorRate": s.error_rate, "rateLimit": s.rate_limit,
         }
 
     @app.post("/__control/settings")
@@ -220,6 +306,18 @@ def create_app(state: MockApiState | None = None) -> FastAPI:
             s.ignore_cursor = bool(payload["ignore_cursor"])
         if "no_sort" in payload:
             s.no_sort = bool(payload["no_sort"])
+        if "load_mode" in payload:
+            s.load_mode = bool(payload["load_mode"])
+        if "page_size" in payload:
+            s.page_size = max(0, int(payload["page_size"]))
+        if "total_rows_limit" in payload:
+            s.total_rows_limit = max(0, int(payload["total_rows_limit"]))
+        if "random_delay" in payload:
+            s.random_delay = max(0.0, float(payload["random_delay"]))
+        if "error_rate" in payload:
+            s.error_rate = min(1.0, max(0.0, float(payload["error_rate"])))
+        if "rate_limit" in payload:
+            s.rate_limit = max(0.0, float(payload["rate_limit"]))
         return get_settings()
 
     @app.post("/__control/overrides")
@@ -240,6 +338,11 @@ def create_app(state: MockApiState | None = None) -> FastAPI:
     def get_audit():
         return {"entries": state.audit[-50:]}
 
+    @app.get("/__control/load-stats")
+    def get_load_stats():
+        """P2 压力模式统计快照（§19：请求量、并发、QPS、耗时和错误）。"""
+        return {"loadMode": state.settings.load_mode, "stats": state.load_stats.snapshot()}
+
     @app.get("/api/{obj:path}")
     def get_object(
         obj: str,
@@ -249,6 +352,10 @@ def create_app(state: MockApiState | None = None) -> FastAPI:
         started = time.time()
         path = request.url.path
         s = state.settings
+        status_code = 200
+        load_on = s.load_mode
+        if load_on:
+            state.load_stats.begin()
         try:
             if s.timeout_seconds > 0:
                 # 超时模拟：睡眠超过 Ai-DOP 读取超时（由测试人员配置秒数）
@@ -260,9 +367,17 @@ def create_app(state: MockApiState | None = None) -> FastAPI:
                                 media_type=s.content_type or "application/json")
             if s.delay_seconds > 0:
                 time.sleep(s.delay_seconds)
+            if s.random_delay > 0:
+                # P2 随机延迟（固定/随机延迟测试类型）
+                time.sleep(random.uniform(0, s.random_delay))
             if s.force_status:
                 raise HTTPException(status_code=s.force_status,
                                     detail=f"forced error {s.force_status}")
+            if load_on and s.rate_limit > 0 and not _rate_allowed(state, s.rate_limit):
+                raise HTTPException(status_code=429,
+                                    detail=f"mock load rate limit {s.rate_limit}/s exceeded")
+            if load_on and s.error_rate > 0 and random.random() < s.error_rate:
+                raise HTTPException(status_code=500, detail="mock load error injection")
             _check_auth(state, request)
             if path in state.overrides:
                 rows = [dict(r) for r in state.overrides[path]]
@@ -276,12 +391,22 @@ def create_app(state: MockApiState | None = None) -> FastAPI:
                 rows.sort(key=lambda r: str(r.get("bizKey", "")))
             if cursor and not s.ignore_cursor:
                 rows = [r for r in rows if str(r.get("bizKey", "")) > cursor]
+            if load_on and s.total_rows_limit > 0:
+                rows = rows[:s.total_rows_limit]
+            if load_on and s.page_size > 0:
+                rows = rows[:s.page_size]
             body = _build_response(rows, s.response_path)
             if s.content_type:
                 return Response(content=json.dumps(body, ensure_ascii=False),
                                 media_type=s.content_type)
             return body
+        except HTTPException as ex:
+            status_code = ex.status_code
+            raise
         finally:
+            elapsed = time.time() - started
+            if load_on:
+                state.load_stats.end(status_code, elapsed)
             _audit(state, {
                 "time": time.strftime("%Y-%m-%dT%H:%M:%S%z"),
                 "path": path,
@@ -289,7 +414,9 @@ def create_app(state: MockApiState | None = None) -> FastAPI:
                 "authMode": state.settings.auth_mode,
                 "forceStatus": state.settings.force_status,
                 "faults": _active_faults(state.settings),
-                "elapsedMs": int((time.time() - started) * 1000),
+                "loadMode": load_on,
+                "httpStatus": status_code,
+                "elapsedMs": int(elapsed * 1000),
                 "headers": redact_headers(dict(request.headers), state.secrets()),
             })
 

tools/integration-simulator/providers/mock_db.py

@@ -10,6 +10,7 @@ from __future__ import annotations
 
 import json
 import os
+import time
 from datetime import datetime
 
 # 允许创建的模拟库名；其余含 aidop 的库名一律阻断
@@ -277,6 +278,106 @@ def table_stats(table: str, increment_column: str = "sourceUpdatedAt") -> dict:
         conn.close()
 
 
+# ---------------------------------------------------------------------------
+# 第三阶段 P1 批量数据生成（§18：档位生成、增量比例、字段宽度、一键停止）
+# ---------------------------------------------------------------------------
+
+# §18 数据规模档位（XL 须审批后自定义行数）
+LOAD_SIZES = {"S": 100, "M": 10_000, "L": 100_000}
+
+
+def _pad_text(value, width: int) -> object:
+    if width <= 0 or not isinstance(value, str):
+        return value
+    if len(value) >= width:
+        return value
+    return value + "x" * (width - len(value))
+
+
+def generate_row_blocks(template_rows: list[dict], count: int, *, increment_ratio: float = 1.0,
+                        field_width: int = 0, batch_size: int = 1000,
+                        stop_event=None):
+    """分块生成压力行（生成器，纯函数可单测）。每块生成前检查 stop_event，
+    支持「一键停止后续数据生成」（§18 模拟器功能）。
+
+    - increment_ratio：新业务键占比（0-1）；其余行复用模板 bizKey，模拟「更新已有行」；
+    - field_width：文本字段垫长到该宽度（模拟大字段）；
+    - 每行 bizKey 唯一后缀、sourceUpdatedAt 递增，保证业务键不重复且增量可推进。
+    """
+    if not template_rows or count <= 0:
+        return
+    ratio = min(1.0, max(0.0, increment_ratio))
+    new_count = int(count * ratio)
+    for start in range(0, count, batch_size):
+        if stop_event is not None and stop_event.is_set():
+            return
+        block = []
+        for i in range(start, min(start + batch_size, count)):
+            row = dict(template_rows[i % len(template_rows)])
+            if i < new_count:
+                if "bizKey" in row:
+                    row["bizKey"] = f"{row['bizKey']}#LOAD-{i}"
+            if field_width > 0:
+                row = {k: _pad_text(v, field_width) for k, v in row.items()}
+            if "sourceUpdatedAt" in row:
+                row["sourceUpdatedAt"] = (datetime.now().strftime("%Y-%m-%dT%H:%M:%S")
+                                          + f".{i % 1000:03d}")
+            block.append(row)
+        yield block
+
+
+def bulk_generate(table: str, template_rows: list[dict], count: int, *, increment_ratio: float = 1.0,
+                  field_width: int = 0, batch_size: int = 1000, stop_event=None) -> dict:
+    """分块装载大档数据（S/M/L）；每块检查 stop_event 支持一键停止。
+    返回生成耗时、实际装载行数、块数、是否被停止。"""
+    started = time.monotonic()
+    loaded = 0
+    batches = 0
+    stopped = False
+    for block in generate_row_blocks(template_rows, count, increment_ratio=increment_ratio,
+                                     field_width=field_width, batch_size=batch_size,
+                                     stop_event=stop_event):
+        if stop_event is not None and stop_event.is_set():
+            stopped = True
+            break
+        upsert_rows(table, block)
+        loaded += len(block)
+        batches += 1
+    return {"table": table, "requested": count, "loaded": loaded, "batches": batches,
+            "stopped": stopped, "elapsedSeconds": round(time.monotonic() - started, 2)}
+
+
+def db_status() -> dict:
+    """模拟库连接与基础统计（§18：查看数据库侧当前连接和基础统计）。"""
+    import pymysql
+    params = mysql_conn_params()
+    database = params["database"]
+    conn = pymysql.connect(cursorclass=pymysql.cursors.DictCursor, autocommit=True, **params)
+    try:
+        with conn.cursor() as cur:
+            cur.execute("SHOW STATUS LIKE 'Threads_connected'")
+            threads = int((cur.fetchone() or {}).get("Value") or 0)
+            cur.execute("SHOW STATUS LIKE 'Uptime'")
+            uptime = int((cur.fetchone() or {}).get("Value") or 0)
+            cur.execute(
+                "SELECT COUNT(*) AS cnt FROM information_schema.tables "
+                "WHERE table_schema=%s", (database,))
+            tables = int((cur.fetchone() or {}).get("cnt") or 0)
+            cur.execute(
+                "SELECT table_name, table_rows FROM information_schema.tables "
+                "WHERE table_schema=%s ORDER BY table_name LIMIT 200", (database,))
+            table_rows = {r["table_name"]: r["table_rows"] for r in cur.fetchall()}
+            cur.execute(
+                "SELECT COUNT(*) AS cnt FROM information_schema.processlist "
+                "WHERE db=%s", (database,))
+            active = int((cur.fetchone() or {}).get("cnt") or 0)
+        return {"database": database, "threadsConnected": threads, "uptimeSeconds": uptime,
+                "tableCount": tables, "tableRows": table_rows,
+                "connectionsOnDatabase": active}
+    finally:
+        conn.close()
+
+
 # ---------------------------------------------------------------------------
 # Ai-DOP 配置清单生成（只输出建议，不写库）
 # ---------------------------------------------------------------------------

tools/integration-simulator/static/app.js

@@ -568,6 +568,24 @@ async function init() {
   $('btn-fault-report-refresh').addEventListener('click', refreshFaultReportList);
   refreshFaultCatalog();
 
+  // ---- 压力测试（第三阶段）----
+  $('load-window').value = defaultTimeWindow(); // 先填默认值，不依赖后续绑定成败
+  fillLoadObjectSelect(null);
+  $('load-object').addEventListener('change', syncLoadEntity);
+  $('load-channel').addEventListener('change', renderLoadParams);
+  const probeBtn = $('btn-entity-probe');
+  if (probeBtn) probeBtn.addEventListener('click', () => fillEntitySelect(true));
+  $('btn-load-plan').addEventListener('click', submitLoadPlan);
+  $('btn-load-start').addEventListener('click', startLoad);
+  $('btn-load-stop').addEventListener('click', stopLoad);
+  $('btn-load-report-save').addEventListener('click', saveLoadReport);
+  $('btn-load-report-export').addEventListener('click', () => exportLoadReport(LAST_LOAD_REPORT_ID));
+  $('btn-load-report-refresh').addEventListener('click', refreshLoadReportList);
+  $('btn-load-checklist').addEventListener('click', renderLoadChecklist);
+  renderLoadParams();
+  loadLimits();
+  setInterval(pollLoadStatus, 2000);
+
   refreshChannelState();
   renderScenarioGraph();
   await refreshStatus();
@@ -578,3 +596,319 @@ init().catch((ex) => {
   $('env-banner').className = 'banner err';
   $('env-banner').textContent = '❌ 初始化失败：' + ex.message;
 });
+
+/* ================= 压力测试（第三阶段 §16-§21） ================= */
+let LAST_LOAD_REPORT_ID = null;
+let LOAD_LIMITS = null;
+
+function loadChannel() { return $('load-channel').value; }
+
+async function loadLimits() {
+  try {
+    const d = await api('/api/load/limits');
+    LOAD_LIMITS = d;
+    const parts = Object.entries(d.defaultLimits)
+      .map(([k, v]) => `${k}≤${v}（硬限 ${d.hardLimits[k]}）`);
+    $('load-limits').textContent = `保守上限：${parts.join('；')}。超出须勾选「提高上限二次确认」，越过硬限不允许。档位：S=${d.sizes.S} M=${d.sizes.M} L=${d.sizes.L}（XL 须审批自定义）`;
+  } catch (ex) { setError('load-limits', ex); }
+}
+
+function defaultTimeWindow(minutes = 15) {
+  // 预填「当前时间起 N 分钟」窗口（YYYY-MM-DD HH:MM-HH:MM），用户可改
+  const fmt = (d) => {
+    const p = (n) => String(n).padStart(2, '0');
+    return `${d.getFullYear()}-${p(d.getMonth() + 1)}-${p(d.getDate())} ${p(d.getHours())}:${p(d.getMinutes())}`;
+  };
+  const start = new Date();
+  const end = new Date(start.getTime() + minutes * 60000);
+  return `${fmt(start)}-${fmt(end)}`;
+}
+
+function entityOptions(ch) {
+  // 按通道生成实体下拉选项：方式丙用入站实体码；方式甲/乙用对象对应的 SIM 实体码
+  if (ch === 'API_INBOUND') {
+    return objects().filter((o) => o.apiInbound && o.apiInbound.supported)
+      .map((o) => ({ value: o.apiInbound.entityCode, label: `${o.apiInbound.entityCode} · ${o.businessName}` }));
+  }
+  if (ch === 'DB_SYNC') {
+    return objects().filter((o) => o.dbSync && o.dbSync.supported)
+      .map((o) => ({ value: `SIM_${o.objectCode}_DB`, label: `SIM_${o.objectCode}_DB · ${o.businessName}` }));
+  }
+  if (ch === 'API_PULL') {
+    return objects().filter((o) => o.apiPull && o.apiPull.supported)
+      .map((o) => ({ value: `SIM_${o.objectCode}_API`, label: `SIM_${o.objectCode}_API · ${o.businessName}` }));
+  }
+  return [];
+}
+
+function fillLoadObjectSelect(authorized) {
+  // 对象下拉按当前通道过滤：方式丙只留「已授权入站对象」，保证选任何对象都能带出实体
+  const ch = loadChannel();
+  const sel = $('load-object');
+  const prev = sel.value;
+  const list = ch === 'API_INBOUND'
+    ? objects().filter((o) => (o.apiInbound || {}).supported &&
+        (!authorized || authorized.has(o.apiInbound.entityCode)))
+    : objects();
+  sel.innerHTML = '';
+  list.forEach((o) => {
+    const opt = document.createElement('option');
+    opt.value = o.objectCode;
+    opt.textContent = `${o.objectCode} · ${o.businessName}`;
+    sel.appendChild(opt);
+  });
+  if (list.length && !Array.from(sel.options).some((o) => o.value === prev)) {
+    sel.value = list[0].objectCode;
+  }
+}
+
+let ENTITY_FILL_TOKEN = 0;
+
+async function fillEntitySelect(force = false) {
+  const token = ++ENTITY_FILL_TOKEN;
+  const ch = loadChannel();
+  const sel = $('load-entity');
+  const opts = entityOptions(ch);
+  if (ch === 'API_INBOUND') {
+    // 方式丙：探测 Ai-DOP 侧授权状态（后端代理，secret 不出后端），只展示已授权实体
+    sel.disabled = true;
+    sel.innerHTML = '<option value="">（探测授权中…）</option>';
+    let authorized = null;
+    try {
+      const d = await api(`/api/inbound/auth-probe${force ? '?refresh=1' : ''}`);
+      if (token !== ENTITY_FILL_TOKEN) return; // 已有更新的填充请求，放弃本次写 UI
+      authorized = new Set(Object.keys(d.entities || {}).filter((e) => d.entities[e]));
+    } catch (ex) {
+      setError('load-limits', ex);
+    }
+    if (authorized) {
+      const ok = opts.filter((o) => authorized.has(o.value));
+      sel.innerHTML = ok.length
+        ? '<option value="">（跟随所选对象）</option>' +
+          ok.map((o) => `<option value="${o.value}">${o.label}</option>`).join('')
+        : '<option value="">（无已授权实体，请先在 Ai-DOP 侧授权）</option>';
+      fillLoadObjectSelect(authorized);
+    } else {
+      // 探测失败：回退全量选项，保证表单可用
+      sel.innerHTML = '<option value="">（探测失败，显示全部）</option>' +
+        opts.map((o) => `<option value="${o.value}">${o.label}</option>`).join('');
+      fillLoadObjectSelect(null);
+    }
+    sel.disabled = false;
+  } else {
+    sel.innerHTML = '<option value="">（跟随所选对象）</option>' +
+      opts.map((o) => `<option value="${o.value}">${o.label}</option>`).join('');
+    sel.disabled = true;
+    fillLoadObjectSelect(null);
+  }
+  syncLoadEntity();
+}
+
+function syncLoadEntity() {
+  // 对象/通道切换后默认选中该对象对应的实体码（方式甲/乙为 SIM 实体码，仅展示）
+  const ch = loadChannel();
+  const obj = findObj($('load-object').value);
+  const val = obj ? {
+    API_INBOUND: (obj.apiInbound || {}).entityCode,
+    DB_SYNC: `SIM_${obj.objectCode}_DB`,
+    API_PULL: `SIM_${obj.objectCode}_API`,
+  }[ch] : '';
+  if (val) {
+    const sel = $('load-entity');
+    sel.value = Array.from(sel.options).some((o) => o.value === val) ? val : '';
+  }
+}
+
+function renderLoadParams() {
+  fillEntitySelect();
+  const ch = loadChannel();
+  const box = $('load-params');
+  const field = (label, inner) => `<label>${label} ${inner}</label>`;
+  if (ch === 'API_INBOUND') {
+    box.innerHTML = [
+      field('模式', `<select id="loadp-mode"><option value="push">单条/多行 POST</option><option value="bulk">bulk（NDJSON）</option></select>`),
+      field('批大小（bulk 每请求行数）', `<input id="loadp-batch" type="number" min="1" value="1" />`),
+      field('预热时间（秒）', `<input id="loadp-warm" type="number" min="0" value="0" />`),
+      field('样例模式', `<select id="loadp-sample"><option value="fixed">固定样例</option><option value="random">随机样例</option></select>`),
+      field('错误数据比例（0-1）', `<input id="loadp-error" type="number" min="0" max="1" step="0.05" value="0" />`),
+      field('幂等重复', `<select id="loadp-idem"><option value="false">否（每请求新幂等键）</option><option value="true">是（复用同幂等键）</option></select>`),
+      field('超时（秒）', `<input id="loadp-timeout" type="number" min="1" value="30" />`),
+      field('自动停止阈值（连续错误/超时/429）',
+        `<input id="loadp-cerr" type="number" min="1" value="5" style="width:4em" /> / ` +
+        `<input id="loadp-tout" type="number" min="1" value="5" style="width:4em" /> / ` +
+        `<input id="loadp-r429" type="number" min="1" value="5" style="width:4em" />`),
+      field('阶梯升压（可选 JSON）', `<textarea id="loadp-ramp" rows="3" placeholder='留空=固定档；如 [{"duration":10,"qps":5,"concurrency":2},{"duration":10,"qps":20,"concurrency":8}]'></textarea>`),
+    ].join('');
+  } else if (ch === 'API_PULL') {
+    box.innerHTML = [
+      field('每次返回行数（0=全部）', `<input id="loadp-page" type="number" min="0" value="0" />`),
+      field('响应数据总量上限（0=不截断）', `<input id="loadp-total" type="number" min="0" value="0" />`),
+      field('随机延迟上限（秒）', `<input id="loadp-rdelay" type="number" min="0" step="0.1" value="0" />`),
+      field('错误比例（0-1，间歇 5xx）', `<input id="loadp-error" type="number" min="0" max="1" step="0.05" value="0" />`),
+      field('限流阈值（每秒最大请求数，0=不限）', `<input id="loadp-rlimit" type="number" min="0" value="0" />`),
+    ].join('');
+  } else {
+    box.innerHTML = [
+      field('数据规模档位', `<select id="loadp-size">` +
+        `<option value="S">S：100 行</option><option value="M">M：10,000 行</option>` +
+        `<option value="L">L：100,000 行</option><option value="XL">XL：自定义（须审批）</option></select>`),
+      field('增量比例（新业务键占比 0-1）', `<input id="loadp-ratio" type="number" min="0" max="1" step="0.1" value="1" />`),
+      field('字段宽度（0=样例原样）', `<input id="loadp-width" type="number" min="0" value="0" />`),
+    ].join('');
+    const sizes = (LOAD_LIMITS && LOAD_LIMITS.sizes) || { S: 100, M: 10000, L: 100000 };
+    $('loadp-size').addEventListener('change', () => {
+      const v = $('loadp-size').value;
+      if (sizes[v] !== undefined) $('load-rows').value = sizes[v];
+    });
+  }
+}
+
+function loadPlanPayload() {
+  const ch = loadChannel();
+  const num = (id, dft = 0) => {
+    const v = parseFloat($(id).value);
+    return Number.isFinite(v) ? v : dft;
+  };
+  const plan = {
+    channel: ch,
+    objectCode: $('load-object').value,
+    entityCode: ch === 'API_INBOUND' ? $('load-entity').value.trim() : '',
+    environment: $('load-env').value.trim(),
+    tenant: $('load-tenant').value.trim(),
+    owner: $('load-owner').value.trim(),
+    timeWindow: $('load-window').value.trim(),
+    maxDataRows: num('load-rows'),
+    maxConcurrency: num('load-conc'),
+    maxQps: num('load-qps'),
+    maxDurationSeconds: num('load-dur'),
+    stopThreshold: $('load-stopth').value.trim(),
+    cleanupPlan: $('load-cleanup').value.trim(),
+    confirmed: $('load-confirmed').checked,
+    escalationConfirm: $('load-escalation').checked,
+  };
+  if (ch === 'API_INBOUND') {
+    Object.assign(plan, {
+      mode: $('loadp-mode').value,
+      batchSize: num('loadp-batch', 1),
+      warmupSeconds: num('loadp-warm'),
+      sampleMode: $('loadp-sample').value,
+      errorRate: num('loadp-error'),
+      idemReplay: $('loadp-idem').value === 'true',
+      timeoutSeconds: num('loadp-timeout', 30),
+      maxConsecutiveErrors: num('loadp-cerr', 5),
+      maxTimeouts: num('loadp-tout', 5),
+      maxRateLimited: num('loadp-r429', 5),
+    });
+    const ramp = $('loadp-ramp').value.trim();
+    if (ramp) {
+      try { plan.rampSteps = JSON.parse(ramp); } catch { plan.rampSteps = null; }
+    }
+  } else if (ch === 'API_PULL') {
+    Object.assign(plan, {
+      pageSize: num('loadp-page'),
+      totalRowsLimit: num('loadp-total'),
+      randomDelay: num('loadp-rdelay'),
+      errorRate: num('loadp-error'),
+      rateLimit: num('loadp-rlimit'),
+    });
+  } else {
+    Object.assign(plan, {
+      size: $('loadp-size').value,
+      incrementRatio: num('loadp-ratio', 1),
+      fieldWidth: num('loadp-width'),
+    });
+  }
+  return plan;
+}
+
+async function submitLoadPlan() {
+  try {
+    const d = await api('/api/load/plan', { method: 'POST', body: JSON.stringify(loadPlanPayload()) });
+    show('load-status', { message: '计划已通过校验（ready），点击「开始压测」显式开启', ...d });
+  } catch (ex) { setError('load-status', ex); }
+}
+
+async function startLoad() {
+  const btn = $('btn-load-start');
+  btn.disabled = true;
+  try {
+    const plan = loadPlanPayload();
+    const d = await api('/api/load/start', { method: 'POST', body: JSON.stringify(plan) });
+    show('load-status', d);
+    await pollLoadStatus();
+  } catch (ex) { setError('load-status', ex); } finally { btn.disabled = false; }
+}
+
+async function stopLoad() {
+  try {
+    const d = await api('/api/load/stop', { method: 'POST', body: JSON.stringify({ reason: 'manual' }) });
+    show('load-status', d);
+    await pollLoadStatus();
+  } catch (ex) { setError('load-status', ex); }
+}
+
+async function pollLoadStatus() {
+  // 只在压力页签激活且存在运行态时轮询，避免无谓请求
+  if (!$('tab-load').classList.contains('active')) return;
+  try {
+    const d = await api('/api/load/status');
+    const { guard, metrics, p2, p1 } = d;
+    const brief = {
+      状态: guard.state,
+      runId: guard.runId || '—',
+      停止原因: guard.stopReason || '—',
+      无残留: guard.noResidual,
+      请求数: metrics.totalRequests,
+      数据行: metrics.totalRows,
+      成功: metrics.success, 失败: metrics.failed,
+      成功率: metrics.successRate,
+      实际QPS: metrics.actualQps,
+      'P50/P95/P99(ms)': `${metrics.p50Ms}/${metrics.p95Ms}/${metrics.p99Ms}`,
+      超时: metrics.timeouts, 限流: metrics.rateLimited,
+      HTTP错误: metrics.httpErrors,
+    };
+    if (p2) Object.assign(brief, { 'P2请求/并发峰值': `${p2.stats.requests}/${p2.stats.peakConcurrency}`, P2限流: p2.stats.rateLimited });
+    if (p1 && p1.result) Object.assign(brief, { P1装载: `${p1.result.loaded}/${p1.result.requested}`, 耗时秒: p1.result.elapsedSeconds, 被停止: p1.result.stopped });
+    if (p1 && p1.error) Object.assign(brief, { P1错误: p1.error });
+    $('load-status').textContent = JSON.stringify(brief, null, 2);
+  } catch { /* 轮询失败静默，下次重试 */ }
+}
+
+async function saveLoadReport() {
+  const btn = $('btn-load-report-save');
+  btn.disabled = true;
+  try {
+    let runId = '';
+    try {
+      const s = await api('/api/load/status');
+      runId = (s.guard && s.guard.runId) || '';
+    } catch { /* 状态不可得时不阻塞报告保存 */ }
+    const d = await api('/api/load/report', {
+      method: 'POST',
+      body: JSON.stringify({
+        runId,
+        channel: loadChannel(),
+        aidopResource: $('load-report-res').value.trim(),
+        aidopFlow: $('load-report-flow').value.trim(),
+        conclusion: $('load-report-concl').value.trim(),
+        notes: $('load-report-notes').value.trim(),
+      }),
+    });
+    LAST_LOAD_REPORT_ID = d.report.id;
+    show('load-report-list', { message: `已保存压力报告 ${d.report.id}（保存前已脱敏）`, report: d.report });
+  } catch (ex) { setError('load-report-list', ex); } finally { btn.disabled = false; }
+}
+
+async function exportLoadReport(id) {
+  if (!id) { setError('load-report-list', '先保存报告再导出'); return; }
+  window.open(`/api/load/report/export/${id}`, '_blank');
+}
+
+async function refreshLoadReportList() {
+  try { show('load-report-list', await api('/api/load/report/list')); } catch (ex) { setError('load-report-list', ex); }
+}
+
+async function renderLoadChecklist() {
+  try { show('load-checklist', await api(`/api/load/aidop-checklist/${loadChannel()}`)); }
+  catch (ex) { setError('load-checklist', ex); }
+}

tools/integration-simulator/static/index.html

@@ -19,6 +19,7 @@
     <button data-tab="tab-db">模拟 MySQL（DB_SYNC）</button>
     <button data-tab="tab-inbound">API_INBOUND</button>
     <button data-tab="tab-fault">异常测试</button>
+    <button data-tab="tab-load">压力测试</button>
     <button data-tab="tab-report">全链路报告</button>
     <button data-tab="tab-runs">运行记录</button>
     <button data-tab="tab-security">环境安全</button>
@@ -242,6 +243,89 @@
     </div>
   </section>
 
+  <!-- 压力测试（第三阶段 §16-§21） -->
+  <section id="tab-load" class="tab">
+    <div class="grid2">
+      <div class="card">
+        <h2>压力计划与安全确认（§16）</h2>
+        <p class="muted">
+          压力测试必须由测试人员显式开启，默认不开启、不允许对生产运行。
+          执行前必须填写全部十项；共享开发/UAT 环境须由负责人确认后执行。
+        </p>
+        <label>压测方式
+          <select id="load-channel">
+            <option value="API_INBOUND" selected>方式丙 API_INBOUND（模拟器发压）</option>
+            <option value="API_PULL">方式乙 API_PULL（Mock 承受拉取压力）</option>
+            <option value="DB_SYNC">方式甲 DB_SYNC（生成档位源数据）</option>
+          </select>
+        </label>
+        <label>对象 <select id="load-object"></select></label>
+        <label>实体
+          <select id="load-entity"></select>
+        </label>
+        <div class="row">
+          <span class="muted">仅显示已授权实体（探测 GET schema，无副作用）；方式甲/乙显示对象对应 SIM 实体码</span>
+          <button id="btn-entity-probe" type="button">刷新授权</button>
+        </div>
+        <label>环境
+          <input id="load-env" type="text" list="load-env-opts" placeholder="如：开发库 / 指定主机（禁止生产）" />
+          <datalist id="load-env-opts">
+            <option value="开发库（dev）"></option>
+            <option value="UAT 环境"></option>
+            <option value="指定主机"></option>
+          </datalist>
+        </label>
+        <label>测试租户 <input id="load-tenant" type="text" placeholder="Ai-DOP 测试租户编号（按实际绑定填写）" /></label>
+        <label>负责人 <input id="load-owner" type="text" placeholder="测试负责人姓名" /></label>
+        <label>时间窗口 <input id="load-window" type="text" placeholder="如：2026-10-13 14:00-14:30（已预填当前 15 分钟）" /></label>
+        <label>最大数据量（行）<input id="load-rows" type="number" min="1" value="1000" /></label>
+        <label>最大并发 <input id="load-conc" type="number" min="1" value="5" /></label>
+        <label>最大 QPS <input id="load-qps" type="number" min="1" value="10" /></label>
+        <label>最大持续时间（秒）<input id="load-dur" type="number" min="1" value="30" /></label>
+        <label>停止阈值 <input id="load-stopth" type="text" value="连续 5 次错误/超时/429 自动停止" /></label>
+        <label>清理方案 <input id="load-cleanup" type="text" value="压测结束后清理模拟器侧测试数据，并核对 Ai-DOP 贴源/流水" /></label>
+        <div class="row">
+          <label class="inline"><input id="load-escalation" type="checkbox" /> 提高上限二次确认</label>
+          <label class="inline"><input id="load-confirmed" type="checkbox" /> 负责人已确认（共享环境）</label>
+        </div>
+        <div id="load-limits" class="muted"></div>
+      </div>
+      <div class="card">
+        <h2>通道参数（§17 / §18 / §19 / §20）</h2>
+        <div id="load-params"></div>
+        <div class="row">
+          <button id="btn-load-plan" class="primary">提交计划（校验）</button>
+          <button id="btn-load-start" class="primary">开始压测</button>
+          <button id="btn-load-stop">立即停止</button>
+        </div>
+        <h3>实时指标与错误摘要</h3>
+        <pre id="load-status" class="output small">默认关闭。提交计划 → 开始后显示实时指标。</pre>
+      </div>
+    </div>
+    <div class="grid2">
+      <div class="card">
+        <h2>压力报告（自动指标 + 人工观察）</h2>
+        <label>Ai-DOP 资源观察（CPU/内存/队列/DB）
+          <textarea id="load-report-res" rows="2" placeholder="在 Ai-DOP 监控工具观察后填写"></textarea></label>
+        <label>Ai-DOP 流水核对
+          <textarea id="load-report-flow" rows="2" placeholder="mdp_sync_log / mdp_inbound_request 核对结果"></textarea></label>
+        <label>结论 <input id="load-report-concl" type="text" /></label>
+        <label>备注 <textarea id="load-report-notes" rows="2"></textarea></label>
+        <div class="row">
+          <button id="btn-load-report-save" class="primary">保存报告（自动脱敏）</button>
+          <button id="btn-load-report-export">导出最新 Markdown</button>
+          <button id="btn-load-report-refresh">刷新列表</button>
+        </div>
+        <pre id="load-report-list" class="output small">—</pre>
+      </div>
+      <div class="card">
+        <h2>Ai-DOP 人工执行与观察清单</h2>
+        <div class="row"><button id="btn-load-checklist">生成清单</button></div>
+        <pre id="load-checklist" class="output small">选择方式后点击「生成清单」</pre>
+      </div>
+    </div>
+  </section>
+
   <!-- 全链路报告 -->
   <section id="tab-report" class="tab">
     <div class="grid2">
@@ -323,6 +407,6 @@
     </div>
   </section>
 
-  <script src="/static/app.js"></script>
+  <script src="/static/app.js?v=20261014b"></script>
 </body>
 </html>

tools/integration-simulator/tests/test_auth_probe.py

@@ -0,0 +1,83 @@
+"""auth-probe 路由测试：入站实体授权探测与过滤（进程内，无网络）。"""
+from __future__ import annotations
+
+import time
+
+import pytest
+
+fastapi_testclient = pytest.importorskip("fastapi.testclient")
+TestClient = fastapi_testclient.TestClient
+
+import app as sim_app  # noqa: E402
+from catalog import sample_adapters as sa  # noqa: E402
+
+_CATALOG = {"objects": [
+    {"objectCode": "ITEM", "apiInbound": {"supported": True,
+                                           "entityCode": "MDM_ITEM",
+                                           "contractVersion": "v1"}},
+    {"objectCode": "CUSTOMER", "apiInbound": {"supported": True,
+                                              "entityCode": "MDM_CUSTOMER",
+                                              "contractVersion": "v1"}},
+    {"objectCode": "NO_INBOUND", "apiInbound": {"supported": False,
+                                                "entityCode": "MDM_X"}},
+]}
+
+
+class _FakeClient:
+    def op_schema(self, entity, contract_version=None):
+        return (403 if entity == "MDM_CUSTOMER" else 200), {"message": "ok"}, {}
+
+
+def _reset_cache(monkeypatch):
+    monkeypatch.setitem(sim_app._AUTH_PROBE_CACHE, "at", 0.0)
+    monkeypatch.setitem(sim_app._AUTH_PROBE_CACHE, "data", None)
+
+
+def _patch_reachable(monkeypatch):
+    monkeypatch.setattr(sim_app.aidop_probe, "check_reachable",
+                        lambda *a, **k: {"reachable": True})
+
+
+def test_auth_probe_filters_unauthorized(monkeypatch):
+    monkeypatch.setattr(sim_app, "_build_inbound_client", lambda: _FakeClient())
+    monkeypatch.setattr(sa, "load_catalog", lambda: _CATALOG)
+    _patch_reachable(monkeypatch)
+    _reset_cache(monkeypatch)
+    r = TestClient(sim_app.app).get("/api/inbound/auth-probe", params={"refresh": 1})
+    assert r.status_code == 200
+    body = r.json()
+    assert body["cached"] is False
+    assert body["entities"] == {"MDM_ITEM": True, "MDM_CUSTOMER": False}
+
+
+def test_auth_probe_unreachable_marks_all_false_without_probing(monkeypatch):
+    monkeypatch.setattr(sim_app, "_build_inbound_client",
+                        lambda: (_ for _ in ()).throw(AssertionError("should not call")))
+    monkeypatch.setattr(sa, "load_catalog", lambda: _CATALOG)
+    monkeypatch.setattr(sim_app.aidop_probe, "check_reachable",
+                        lambda *a, **k: {"reachable": False, "error": "down"})
+    _reset_cache(monkeypatch)
+    r = TestClient(sim_app.app).get("/api/inbound/auth-probe", params={"refresh": 1})
+    assert r.status_code == 200
+    body = r.json()
+    assert body["entities"] == {"MDM_ITEM": False, "MDM_CUSTOMER": False}
+    assert any("不可达" in e for e in body["probeErrors"])
+
+
+def test_auth_probe_missing_credentials_400(monkeypatch):
+    monkeypatch.setattr(sim_app, "_env", lambda name: "")
+    _reset_cache(monkeypatch)
+    r = TestClient(sim_app.app).get("/api/inbound/auth-probe", params={"refresh": 1})
+    assert r.status_code == 400
+    assert "AIDOP_SIM_INBOUND_ACCESS_KEY" in r.json()["detail"]
+
+
+def test_auth_probe_serves_cache_without_client(monkeypatch):
+    monkeypatch.setitem(sim_app._AUTH_PROBE_CACHE, "at", time.monotonic())
+    monkeypatch.setitem(sim_app._AUTH_PROBE_CACHE, "data",
+                        {"entities": {"MDM_ITEM": True}, "probedAt": "x", "probeErrors": []})
+    monkeypatch.setattr(sim_app, "_build_inbound_client",
+                        lambda: (_ for _ in ()).throw(AssertionError("should not call")))
+    r = TestClient(sim_app.app).get("/api/inbound/auth-probe")
+    assert r.status_code == 200
+    assert r.json()["cached"] is True

tools/integration-simulator/tests/test_load_controls.py

@@ -0,0 +1,311 @@
+"""第三阶段压力测试安全控制与 P3 执行器测试（任务书 §16/§20/§21 验收表）。"""
+from __future__ import annotations
+
+import threading
+import time
+
+import pytest
+
+from catalog import sample_adapters as sa
+from load import controls
+from load.metrics import LoadMetrics
+from load.runner_p3 import P3Runner, _RateLimiter
+
+
+def _valid_plan(**over) -> dict:
+    plan = {
+        "channel": "API_INBOUND",
+        "environment": "dev", "tenant": "T1", "owner": "qa", "timeWindow": "t",
+        "maxDataRows": 1000, "maxConcurrency": 4, "maxQps": 10, "maxDurationSeconds": 5,
+        "stopThreshold": "连续 5 次错误/超时/429 自动停止", "cleanupPlan": "clean",
+        "confirmed": True,
+    }
+    plan.update(over)
+    return plan
+
+
+# ---------------------------------------------------------------------------
+# validate_plan（§16 十项必填 / 上限 / 二次确认）
+# ---------------------------------------------------------------------------
+
+def test_validate_plan_accepts_complete_plan():
+    assert controls.validate_plan(_valid_plan()) == []
+
+
+def test_validate_plan_rejects_missing_required_fields():
+    errors = controls.validate_plan({"channel": "API_INBOUND"})
+    assert len(errors) >= 10
+    assert any("environment" in e for e in errors)
+
+
+def test_validate_plan_requires_owner_confirmation():
+    errors = controls.validate_plan(_valid_plan(confirmed=False))
+    assert any("负责人确认" in e for e in errors)
+
+
+def test_validate_plan_blocks_over_limit_without_escalation():
+    errors = controls.validate_plan(_valid_plan(maxQps=999))
+    assert any("超过保守上限" in e for e in errors)
+
+
+def test_validate_plan_allows_escalation_but_blocks_hard_limit():
+    ok = controls.validate_plan(_valid_plan(maxQps=60, escalationConfirm=True))
+    assert ok == []
+    errors = controls.validate_plan(_valid_plan(maxQps=10000, escalationConfirm=True))
+    assert any("硬上限" in e for e in errors)
+
+
+def test_validate_plan_rejects_bad_channel_and_error_rate_range():
+    errors = controls.validate_plan(_valid_plan(channel="NOPE"))
+    assert any("channel" in e for e in errors)
+    errors = controls.validate_plan(_valid_plan(errorRate=1.5))
+    assert any("errorRate" in e for e in errors)
+
+
+# ---------------------------------------------------------------------------
+# 租户字段剔除（§20：请求体不得携带任意 tenantId）
+# ---------------------------------------------------------------------------
+
+def test_strip_tenant_fields_removes_tenant_keys():
+    row = {"ItemNum": "A", "tenantId": "999", "tenant_id": "888",
+           "Domain": "SIM", "bizKey": "K1"}
+    stripped = controls.strip_tenant_fields(dict(row))
+    assert "tenantId" not in stripped and "tenant_id" not in stripped
+    assert stripped["Domain"] == "SIM" and stripped["bizKey"] == "K1"
+
+
+# ---------------------------------------------------------------------------
+# AutoStopTracker（§20 自动停止阈值）
+# ---------------------------------------------------------------------------
+
+def test_auto_stop_tracker_consecutive_errors():
+    tracker = controls.AutoStopTracker(consecutive_errors=3)
+    assert not tracker.record(202)
+    assert not tracker.record(500)
+    assert not tracker.record(500)
+    assert tracker.record(500)
+    assert tracker.tripped_by == "consecutiveErrors"
+
+
+def test_auto_stop_tracker_timeouts_and_429():
+    tracker = controls.AutoStopTracker(timeouts=2, rate_limited=2)
+    tracker.record(0)
+    assert tracker.record(0) and tracker.tripped_by == "timeouts"
+    tracker2 = controls.AutoStopTracker(timeouts=2, rate_limited=2)
+    tracker2.record(429)
+    assert tracker2.record(429) and tracker2.tripped_by == "rateLimited"
+
+
+def test_auto_stop_tracker_success_resets_consecutive():
+    tracker = controls.AutoStopTracker(consecutive_errors=3)
+    tracker.record(500)
+    tracker.record(500)
+    tracker.record(202)
+    assert not tracker.record(500)
+    assert not tracker.record(500)
+
+
+# ---------------------------------------------------------------------------
+# LoadGuard 生命周期（默认关闭 / 立即停止 / 无残留）
+# ---------------------------------------------------------------------------
+
+def test_guard_default_state_is_idle_and_blocks_start():
+    guard = controls.LoadGuard()
+    assert guard.state == "idle"
+    with pytest.raises(controls.ValidationError):
+        guard.start()
+    assert guard.snapshot()["noResidual"] is True
+
+
+def test_guard_lifecycle_ready_running_stopping_stopped():
+    guard = controls.LoadGuard()
+    assert guard.submit_plan(_valid_plan())["ok"]
+    assert guard.state == "ready"
+    run_id = guard.start()
+    assert guard.state == "running" and run_id.startswith("LOAD-")
+    assert guard.request_stop("manual")["state"] == "stopping"
+    result = guard.finish()
+    assert result["ok"] and result["residualThreads"] == []
+    assert guard.state == "stopped"
+
+
+def test_guard_finish_reports_residual_worker():
+    guard = controls.LoadGuard()
+    guard.submit_plan(_valid_plan())
+    guard.start()
+    worker = threading.Thread(target=lambda: time.sleep(0.2), daemon=True)
+    worker.start()
+    guard.register_worker(worker)
+    guard.request_stop()
+    result = guard.finish()
+    assert not result["ok"] and "residualThreads" in result
+    worker.join(timeout=2)
+
+
+def test_guard_finish_excludes_current_thread_from_residual():
+    # P1/P3 执行器线程注册自身为 worker 并在自身线程内 finish()，
+    # 不得把当前线程误报为残留（回归：曾导致每次收尾都报 residualThreads）
+    guard = controls.LoadGuard()
+    guard.submit_plan(_valid_plan())
+    guard.start()
+    result = {}
+
+    def _finish_in_worker():
+        guard.register_worker(threading.current_thread())
+        guard.request_stop()
+        result.update(guard.finish())
+
+    worker = threading.Thread(target=_finish_in_worker, daemon=True)
+    worker.start()
+    worker.join(timeout=5)
+    assert result["ok"] is True and result["residualThreads"] == []
+    assert guard.state == "stopped"
+
+
+# ---------------------------------------------------------------------------
+# P3 执行器（固定 QPS 短跑 / 立即停止无残留 / 自动停止 / 错误比例注入）
+# ---------------------------------------------------------------------------
+
+class _FakeInboundClient:
+    """进程内假客户端：捕获调用并返回固定状态，不触网。"""
+
+    def __init__(self, status: int = 202):
+        self.status = status
+        self.calls: list[tuple] = []
+
+    def op_push(self, entity, envelope, **kwargs):
+        self.calls.append(("push", entity, envelope, kwargs))
+        return self.status, {"ok": True}, {}
+
+    def op_bulk(self, entity, lines, **kwargs):
+        self.calls.append(("bulk", entity, lines, kwargs))
+        return self.status, {"ok": True}, {}
+
+
+def _p3_env(plan_over: dict | None = None, client_status: int = 202):
+    guard = controls.LoadGuard()
+    metrics = LoadMetrics()
+    client = _FakeInboundClient(status=client_status)
+    obj = next(o for o in sa.load_catalog()["objects"]
+               if (o.get("apiInbound") or {}).get("supported"))
+    entity = obj["apiInbound"]["entityCode"]
+    rows = sa.adapt_inbound_rows(obj, "T-LOAD")["rows"]
+    plan = _valid_plan(**{"entityCode": entity, "runId": "T-LOAD"})
+    plan.update(plan_over or {})
+    return guard, metrics, client, entity, rows, plan
+
+
+def test_p3_runner_short_run_records_metrics_and_finishes_clean():
+    guard, metrics, client, entity, rows, plan = _p3_env(
+        {"maxDurationSeconds": 1, "maxQps": 30, "maxConcurrency": 4,
+         "maxConsecutiveErrors": 5, "maxTimeouts": 5, "maxRateLimited": 5})
+    guard.submit_plan(plan)
+    guard.start()
+    runner = P3Runner(client, entity, rows, plan, guard, metrics)
+    runner.start()
+    runner.join(timeout=10)
+    assert not runner.is_alive()
+    assert guard.state == "stopped"
+    assert guard.snapshot()["noResidual"] is True
+    snap = metrics.snapshot()
+    assert snap["totalRequests"] > 0
+    assert snap["success"] == snap["totalRequests"]
+    assert snap["successRate"] == 1.0
+    assert snap["p50Ms"] >= 0 and snap["startedAt"] and snap["finishedAt"]
+
+
+def test_p3_runner_immediate_stop_leaves_no_residual():
+    guard, metrics, client, entity, rows, plan = _p3_env(
+        {"maxDurationSeconds": 10, "maxQps": 5, "maxConcurrency": 2})
+    guard.submit_plan(plan)
+    guard.start()
+    runner = P3Runner(client, entity, rows, plan, guard, metrics)
+    runner.start()
+    time.sleep(0.2)
+    guard.request_stop("manual stop")
+    runner.join(timeout=10)
+    assert guard.state == "stopped"
+    assert guard.snapshot()["noResidual"] is True
+    assert metrics.snapshot()["totalRequests"] < 100
+
+
+def test_p3_runner_stop_cancels_queued_futures():
+    # 回归：stop 后只等在途请求，排队未启动的任务必须取消（否则慢请求下收尾拖沓）
+    guard, metrics, client, entity, rows, plan = _p3_env(
+        {"maxDurationSeconds": 30, "maxQps": 50, "maxConcurrency": 2})
+    guard.submit_plan(plan)
+    guard.start()
+
+    def slow_push(entity, envelope, **kwargs):
+        time.sleep(0.5)
+        client.calls.append(("push", entity, envelope, kwargs))
+        return 202, {"ok": True}, {}
+
+    client.op_push = slow_push
+    runner = P3Runner(client, entity, rows, plan, guard, metrics)
+    runner.start()
+    time.sleep(1.0)
+    guard.request_stop("manual stop")
+    started = time.monotonic()
+    runner.join(timeout=10)
+    elapsed = time.monotonic() - started
+    assert not runner.is_alive()
+    assert guard.state == "stopped"
+    assert elapsed < 5, f"停止收尾过慢：{elapsed:.1f}s（排队任务未被取消？）"
+    assert guard.snapshot()["noResidual"] is True
+
+
+def test_p3_runner_auto_stops_on_consecutive_errors():
+    guard, metrics, client, entity, rows, plan = _p3_env(
+        {"maxDurationSeconds": 30, "maxQps": 50, "maxConcurrency": 2,
+         "maxConsecutiveErrors": 2},
+        client_status=500)
+    guard.submit_plan(plan)
+    guard.start()
+    runner = P3Runner(client, entity, rows, plan, guard, metrics)
+    runner.start()
+    runner.join(timeout=10)
+    assert "自动停止" in (guard.stop_reason or "")
+    assert guard.state == "stopped"
+    assert metrics.snapshot()["failed"] >= 2
+
+
+def test_p3_runner_error_rate_injects_missing_required_field():
+    # errorRate 硬上限 0.9；提交计划后改回 1.0 保证每行都缺必填字段（确定性断言注入逻辑）
+    guard, metrics, client, entity, rows, plan = _p3_env(
+        {"maxDurationSeconds": 1, "maxQps": 10, "maxConcurrency": 2,
+         "errorRate": 0.9, "escalationConfirm": True})
+    guard.submit_plan(plan)
+    plan["errorRate"] = 1.0
+    guard.start()
+    runner = P3Runner(client, entity, rows, plan, guard, metrics)
+    runner.start()
+    runner.join(timeout=10)
+    assert client.calls, "压测应发出请求"
+    first_required = runner._first_required
+    for call in client.calls[:3]:
+        envelope = call[2]
+        assert first_required not in envelope["data"]["list"][0]
+
+
+def test_p3_runner_never_injects_tenant_fields():
+    guard, metrics, client, entity, rows, plan = _p3_env(
+        {"maxDurationSeconds": 1, "maxQps": 10, "maxConcurrency": 2})
+    guard.submit_plan(plan)
+    guard.start()
+    runner = P3Runner(client, entity, rows, plan, guard, metrics)
+    runner.start()
+    runner.join(timeout=10)
+    for call in client.calls:
+        row = call[2]["data"]["list"][0]
+        assert "tenantId" not in row and "tenant_id" not in row
+
+
+def test_rate_limiter_caps_qps_approximately():
+    limiter = _RateLimiter(qps=100)
+    started = time.monotonic()
+    for _ in range(5):
+        limiter.wait()
+    elapsed = time.monotonic() - started
+    # 5 个槽位应至少占 4 个间隔（4*10ms），宽松断言避免 CI 抖动误报
+    assert elapsed >= 0.03

tools/integration-simulator/tests/test_load_metrics.py

@@ -0,0 +1,272 @@
+"""第三阶段压力指标、P1 批量生成、P2 压力模式测试（任务书 §17/§18/§19）。
+
+全部进程内验证（纯函数/TestClient），不依赖 Ai-DOP 在线与真实 MySQL。
+"""
+from __future__ import annotations
+
+import threading
+import time
+from concurrent.futures import ThreadPoolExecutor
+
+import pytest
+
+fastapi_testclient = pytest.importorskip("fastapi.testclient")
+TestClient = fastapi_testclient.TestClient
+
+from load import metrics as lm  # noqa: E402
+from providers import mock_api  # noqa: E402
+from providers import mock_db  # noqa: E402
+
+
+# ---------------------------------------------------------------------------
+# percentile（§17 P50/P95/P99 计算）
+# ---------------------------------------------------------------------------
+
+def test_percentile_linear_interpolation():
+    assert lm.percentile([1.0, 2.0, 3.0, 4.0], 50) == 2.5
+    assert lm.percentile([1.0, 2.0, 3.0, 4.0], 25) == 1.75
+
+
+def test_percentile_boundaries_and_empty():
+    assert lm.percentile([1.0, 2.0, 3.0], 0) == 1.0
+    assert lm.percentile([1.0, 2.0, 3.0], 100) == 3.0
+    assert lm.percentile([], 50) == 0.0
+    assert lm.percentile([5.0], 95) == 5.0
+
+
+def test_percentile_p95_p99_of_100_values():
+    vals = [float(i) for i in range(100)]
+    assert lm.percentile(vals, 95) == 94.05
+    assert lm.percentile(vals, 99) == 98.01
+
+
+# ---------------------------------------------------------------------------
+# LoadMetrics（§17 指标采集与快照）
+# ---------------------------------------------------------------------------
+
+def test_record_classifies_statuses_and_snapshot():
+    m = lm.LoadMetrics()
+    m.mark_start()
+    m.record(200, 0.01, rows=1)
+    m.record(201, 0.02)
+    m.record(202, 0.03, batches=2)
+    m.record(500, 0.04)
+    m.record(0, 0.05)
+    m.record(429, 0.06)
+    m.mark_finish()
+    s = m.snapshot()
+    assert s["totalRequests"] == 6
+    assert s["success"] == 3 and s["failed"] == 3
+    assert s["successRate"] == 0.5
+    assert s["timeouts"] == 1 and s["rateLimited"] == 1
+    assert s["httpErrors"] == {500: 1}
+    assert s["totalBatches"] == 7
+    assert s["totalRows"] == 1
+    assert s["avgMs"] == 35.0
+    assert s["maxMs"] == 60.0
+    assert s["p50Ms"] == 35.0
+    assert s["firstFailureAt"] is not None
+    assert s["startedAt"] is not None and s["finishedAt"] is not None
+
+
+def test_snapshot_contains_all_metric_fields():
+    m = lm.LoadMetrics()
+    m.mark_start()
+    m.record(200, 0.1, rows=2)
+    m.mark_finish()
+    s = m.snapshot()
+    for field, _ in lm.METRIC_FIELDS:
+        assert field in s, f"missing metric field: {field}"
+    assert s["totalRequests"] == 1 and s["successRate"] == 1.0
+
+
+def test_actual_qps_empty_and_completion_window_pruning():
+    m = lm.LoadMetrics(window_seconds=10)
+    assert m.actual_qps() == 0.0
+    now = time.monotonic()
+    with m._lock:
+        m._completion_times[:] = [now - 100, now - 5, now - 0.5]
+        pruned = m._pruned_completion_times(now)
+    assert pruned == [now - 5, now - 0.5]
+
+
+def test_row_throughput_uses_wall_clock_span():
+    m = lm.LoadMetrics()
+    m.mark_start()
+    m.record(200, 0.001, rows=10)
+    m.mark_finish()
+    with m._lock:
+        m.started_at = "2026-01-01T10:00:00"
+        m.finished_at = "2026-01-01T10:00:10"
+    assert m.row_throughput() == 1.0
+
+
+def test_snapshot_freezes_qps_after_finish():
+    # 停止后快照的 QPS 必须定格：滚动窗口分母不再随时间变大（用户报告“停止后 QPS 还在变”）
+    m = lm.LoadMetrics(window_seconds=10)
+    m.mark_start()
+    m.record(200, 0.01)
+    m.record(200, 0.01)
+    m.mark_finish()
+    s1 = m.snapshot()
+    time.sleep(0.15)
+    s2 = m.snapshot()
+    assert s1["actualQps"] > 0
+    assert s1["actualQps"] == s2["actualQps"]
+    assert s1["rowThroughput"] == s2["rowThroughput"]
+
+
+def test_mark_start_resets_previous_round_metrics():
+    # 单例跨轮复用：新一轮 mark_start 必须清空上一轮累计（防止多轮指标叠加）
+    m = lm.LoadMetrics()
+    m.mark_start()
+    m.record(200, 0.01, rows=3)
+    m.mark_finish()
+    m.mark_start()
+    s = m.snapshot()
+    assert s["totalRequests"] == 0
+    assert s["totalRows"] == 0
+    assert s["startedAt"] is not None
+    assert s["finishedAt"] is None
+
+
+def test_metrics_thread_safe_concurrent_records():
+    m = lm.LoadMetrics()
+
+    def hit(i: int) -> None:
+        m.record(200 if i % 3 else 500, 0.001)
+
+    with ThreadPoolExecutor(max_workers=8) as pool:
+        list(pool.map(hit, range(300)))
+    s = m.snapshot()
+    assert s["totalRequests"] == 300
+    assert s["success"] == 200 and s["failed"] == 100
+    assert s["httpErrors"] == {500: 100}
+
+
+# ---------------------------------------------------------------------------
+# P1 批量生成 generate_row_blocks（§18：档位/增量比例/字段宽度/一键停止）
+# ---------------------------------------------------------------------------
+
+def test_p1_row_blocks_chunked_by_batch_size():
+    rows = [{"bizKey": "B", "sourceUpdatedAt": "2026-01-01T00:00:00"}]
+    blocks = list(mock_db.generate_row_blocks(rows, 2500, batch_size=1000))
+    assert [len(b) for b in blocks] == [1000, 1000, 500]
+
+
+def test_p1_row_blocks_bizkey_unique_with_full_increment():
+    rows = [{"bizKey": "B"}]
+    blocks = list(mock_db.generate_row_blocks(rows, 50, increment_ratio=1.0, batch_size=10))
+    keys = [r["bizKey"] for b in blocks for r in b]
+    assert len(set(keys)) == 50
+    assert all(k.startswith("B#LOAD-") for k in keys)
+
+
+def test_p1_row_blocks_increment_ratio_reuses_template_keys():
+    rows = [{"bizKey": "B"}]
+    blocks = list(mock_db.generate_row_blocks(rows, 10, increment_ratio=0.5, batch_size=100))
+    keys = [r["bizKey"] for b in blocks for r in b]
+    assert keys[:5] == [f"B#LOAD-{i}" for i in range(5)]
+    assert keys[5:] == ["B"] * 5
+
+
+def test_p1_row_blocks_pads_text_fields():
+    rows = [{"bizKey": "B", "note": "hi"}]
+    (block,) = list(mock_db.generate_row_blocks(rows, 1, field_width=8, batch_size=10))
+    row = block[0]
+    assert row["note"] == "hixxxxxx"
+    assert len(row["bizKey"]) >= 8
+
+
+def test_p1_row_blocks_increment_column_advances():
+    rows = [{"bizKey": "B", "sourceUpdatedAt": "2026-01-01T00:00:00"}]
+    blocks = list(mock_db.generate_row_blocks(rows, 100, batch_size=30))
+    vals = [r["sourceUpdatedAt"] for b in blocks for r in b]
+    assert len(set(vals)) == 100
+
+
+def test_p1_row_blocks_stop_event_before_and_midway():
+    rows = [{"bizKey": "B"}]
+    stop = threading.Event()
+    stop.set()
+    assert list(mock_db.generate_row_blocks(rows, 100, stop_event=stop)) == []
+    stop2 = threading.Event()
+    gen = mock_db.generate_row_blocks(rows, 100, batch_size=10, stop_event=stop2)
+    first = next(gen)
+    stop2.set()
+    assert len(first) == 10
+    assert list(gen) == []
+
+
+# ---------------------------------------------------------------------------
+# P2 压力模式（§19：分页截断/错误比例/限流/统计/恢复）
+# ---------------------------------------------------------------------------
+
+@pytest.fixture()
+def p2client():
+    state = mock_api.MockApiState()
+    state.settings.auth_mode = "NONE"
+    return TestClient(mock_api.create_app(state)), state
+
+
+def _override(c, path: str, n: int) -> None:
+    c.post("/__control/overrides",
+           json={"path": path, "rows": [{"bizKey": f"K{i}", "v": i} for i in range(n)]})
+
+
+def test_p2_page_size_applies_only_in_load_mode(p2client):
+    c, _ = p2client
+    _override(c, "/api/load-a", 10)
+    c.post("/__control/settings", json={"page_size": 3})
+    assert len(c.get("/api/load-a").json()["data"]["list"]) == 10  # 未开压力模式不截断
+    c.post("/__control/settings", json={"load_mode": True})
+    assert len(c.get("/api/load-a").json()["data"]["list"]) == 3
+
+
+def test_p2_total_rows_limit_truncates_in_load_mode(p2client):
+    c, _ = p2client
+    _override(c, "/api/load-b", 10)
+    c.post("/__control/settings", json={"load_mode": True, "total_rows_limit": 5})
+    assert len(c.get("/api/load-b").json()["data"]["list"]) == 5
+
+
+def test_p2_error_rate_injects_500(p2client):
+    c, _ = p2client
+    _override(c, "/api/load-c", 3)
+    c.post("/__control/settings", json={"load_mode": True, "error_rate": 1.0})
+    assert c.get("/api/load-c").status_code == 500
+
+
+def test_p2_rate_limit_returns_429_beyond_threshold(p2client):
+    c, _ = p2client
+    _override(c, "/api/load-d", 3)
+    c.post("/__control/settings", json={"load_mode": True, "rate_limit": 1})
+    assert c.get("/api/load-d").status_code == 200
+    assert c.get("/api/load-d").status_code == 429
+
+
+def test_p2_load_stats_tracks_requests_and_errors(p2client):
+    c, _ = p2client
+    _override(c, "/api/load-e", 3)
+    c.post("/__control/settings", json={"load_mode": True, "error_rate": 1.0})
+    c.get("/api/load-e")
+    stats = c.get("/__control/load-stats").json()
+    assert stats["loadMode"] is True
+    assert stats["stats"]["requests"] == 1
+    # JSON 序列化后状态码键为字符串
+    assert stats["stats"]["statusCounts"] == {"500": 1}
+    assert set(stats["stats"]) >= {
+        "requests", "active", "peakConcurrency", "qpsWindow",
+        "avgMs", "maxMs", "statusCounts", "rateLimited"}
+
+
+def test_p2_reset_clears_pressure_settings(p2client):
+    c, _ = p2client
+    c.post("/__control/settings", json={
+        "load_mode": True, "page_size": 2, "total_rows_limit": 4,
+        "random_delay": 1.5, "error_rate": 0.5, "rate_limit": 10})
+    c.post("/__control/settings", json={"reset": True})
+    s = c.get("/__control/settings").json()
+    assert s["loadMode"] is False
+    assert s["pageSize"] == 0 and s["totalRowsLimit"] == 0
+    assert s["randomDelay"] == 0.0 and s["errorRate"] == 0.0 and s["rateLimit"] == 0.0