feat(simulator): 第二阶段异常测试能力N1-N4与真实联调缺陷修复

- N1-N3: 三通道43异常场景目录(fault_catalog)、预置/一键恢复、入站17异常用例
- N4: 八列脱敏报告 save/list/export/delete，同场景重复保存原位更新(不产生重复行)
- 缺陷修复: expectedReject仅限Ai-DOP预检409、Mock下拉状态同步、导出下载、partial-reject/bulk复制行保证1好1坏、wrong-field-type改用结构敌对值、报告保存防重复点击
- 测试基线 124 passed, 1 skipped; N1-N4真实联调(远端MySQL+在线Ai-DOP)三层取证通过

tools/integration-simulator/app.py

@@ -12,6 +12,7 @@
 from __future__ import annotations
 
 import asyncio
+import json
 import os
 import socket
 import sys
@@ -27,13 +28,13 @@ import uvicorn
 from fastapi import FastAPI, HTTPException
 from fastapi.responses import FileResponse, Response
 from fastapi.staticfiles import StaticFiles
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
 
 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 providers import mock_api, mock_db
+from providers import fault_catalog, mock_api, mock_db
 
 DEFAULT_CONTROL_PORT = 8017
 MAX_RUNS = 200
@@ -238,6 +239,24 @@ class ReportPayload(BaseModel):
     cleanup: str = ""                 # 清理提示
 
 
+class FaultApplyPayload(BaseModel):
+    scenarioId: str
+    objectCode: str | None = None     # DB/API 场景默认 ITEM
+    entityCode: str | None = None     # N3-15 未启用 entity 场景的目标实体码
+
+
+class FaultReportPayload(BaseModel):
+    """异常测试报告（任务书 §15 八列）。「是否符合预期」JSON 键为 pass。"""
+    scenarioId: str
+    simBehavior: str = ""
+    expected: str = ""
+    actualResponse: str = ""
+    aidopFlowStatus: str = ""
+    targetChanged: str = ""
+    conform: str = Field(default="", alias="pass")
+    note: str = ""
+
+
 CHANNELS = ("DB_SYNC", "API_PULL", "API_INBOUND")
 CHANNEL_LABELS = {
     "DB_SYNC": "方式甲 DB_SYNC（数据库同步）",
@@ -321,6 +340,53 @@ def build_report_markdown(entry: dict) -> str:
 REPORTS = ReportStore()
 
 
+class FaultReportStore(ReportStore):
+    """异常测试报告存储（任务书 §15 N4）。复用脱敏与内存态约定。
+
+    每场景一行最终结论：同 scenarioId 重复保存 = 原位更新（保留原 id），
+    避免误点/重复点保存产生重复行。
+    """
+
+    def save(self, fields: dict, secrets: list[str] | None = None,
+             access_keys: list[str] | None = None) -> dict:
+        scenario_id = fields.get("scenarioId")
+        existing = next((r for r in self.items if r.get("scenarioId") == scenario_id), None)
+        if existing is None:
+            entry = super().save(fields, secrets, access_keys)
+            replaced = False
+        else:
+            entry = existing
+            entry.update(redact_obj(dict(fields), secrets, access_keys))
+            entry["updatedAt"] = time.strftime("%Y-%m-%dT%H:%M:%S")
+            replaced = True
+        scenario = fault_catalog.find_scenario(entry.get("scenarioId", ""))
+        entry["scenarioTitle"] = scenario["title"] if scenario else ""
+        entry["channel"] = scenario["channel"] if scenario else ""
+        entry["replaced"] = replaced
+        return entry
+
+
+FAULT_REPORTS = FaultReportStore()
+
+
+def build_fault_report_markdown(entry: dict) -> str:
+    """异常报告导出 Markdown（任务书 §15 八列 + 脱敏声明）。"""
+    lines = ["# Ai-DOP 第三方系统模拟器 · 异常测试报告", ""]
+    lines.append(f"- 场景：{entry.get('scenarioId', '')} {entry.get('scenarioTitle', '')}")
+    lines.append(f"- 集成方式：{fault_catalog.FAULT_CHANNEL_LABELS.get(entry.get('channel', ''), entry.get('channel', ''))}")
+    lines.append(f"- 创建时间：{entry.get('createdAt', '')}")
+    lines.append("")
+    lines.append("| 项目 | 内容 |")
+    lines.append("|---|---|")
+    for key, label in fault_catalog.FAULT_REPORT_FIELDS:
+        value = str(entry.get(key) or "").strip() or "（未填写）"
+        lines.append(f"| {label} | {value.replace(chr(10), '<br>')} |")
+    lines.append("")
+    lines.append("> 脱敏说明：本报告已自动脱敏，AccessKey 仅保留后四位；"
+                 "数据库密码、Secret、JWT、完整 Authorization 与连接串已遮蔽。")
+    return "\n".join(lines)
+
+
 # ---------------------------------------------------------------------------
 # 公共辅助
 # ---------------------------------------------------------------------------
@@ -374,6 +440,15 @@ def _require_obj_by_entity(entity_code: str) -> dict | None:
     return None
 
 
+def _first_required_field(entity: str) -> str:
+    """契约实体第一个必填字段（N3 缺必填字段/类型错误/部分行拒绝场景用）。"""
+    spec = sa.INBOUND_SPECS.get(entity) or []
+    for field, required, _ in spec:
+        if required:
+            return field
+    return "Descr"
+
+
 def _layers(code: str, config: str, data: str) -> dict:
     return {"code": code, "config": config, "data": data}
 
@@ -739,6 +814,31 @@ def inbound_exec(payload: InboundExecPayload):
             env2 = sa.inbound_envelope([{**r, "sourceUpdatedAt": sa.now_iso()} for r in (rows or [])])
             (s1, d1), (s2, d2) = client.case_same_key_diff_body(entity, envelope, env2)
             status, body, headers = s2, {"first": [s1, d1], "conflict": [s2, d2]}, {}
+        # ---- 第二阶段 N3 新增用例（被 Ai-DOP 正确拒绝 = 测试通过）----
+        elif case == "missing-access-key":
+            status, body, headers = client.case_missing_access_key(entity, envelope)
+        elif case == "wrong-access-key":
+            status, body, headers = client.case_wrong_access_key(entity, envelope)
+        elif case == "missing-required-field":
+            status, body, headers = client.case_missing_required_field(
+                entity, envelope, _first_required_field(entity))
+        elif case == "wrong-field-type":
+            # 用结构敌对值（嵌套对象）而非同类标量：字符串列塞数字可能被宽容转换，
+            # 必须用无法映射到标量列的值才能触发类型校验拒绝
+            status, body, headers = client.case_wrong_field_type(
+                entity, envelope, _first_required_field(entity), {"__sim_type_fault__": True})
+        elif case == "partial-reject":
+            env = json.loads(json.dumps(envelope, ensure_ascii=False))
+            if len(env["data"]["list"]) < 2:
+                # 样例仅 1 行时复制一行，保证「1 好 1 坏」部分行拒绝可演示
+                env["data"]["list"].append(
+                    json.loads(json.dumps(env["data"]["list"][0], ensure_ascii=False)))
+            env["data"]["list"][1].pop(_first_required_field(entity), None)
+            status, body, headers = client.op_push(entity, env, contract_version=contract_version)
+        elif case == "stale":
+            status, body, headers = client.case_stale_rows(entity, envelope)
+        elif case == "unknown-entity":
+            status, body, headers = client.case_unknown_entity(envelope)
         else:
             raise HTTPException(status_code=400, detail=f"unknown case: {case}")
         data_state = "ACCEPTED" if status == 202 else ("RUN_FAILED" if status >= 400 else "NOT_RUN")
@@ -757,6 +857,16 @@ def inbound_exec(payload: InboundExecPayload):
         # 后端 NDJSON 分块：每行须为完整信封（data.list）+ 根级 idempotencyKey（MdpInboundReceiveService.BulkAsync）
         bulk_rows = [{"idempotencyKey": f"{run_id}-L{i + 1}",
                       "data": {"list": [r]}} for i, r in enumerate(rows or [])]
+        if payload.case == "bulk-partial-failure":
+            # N3-16：坏行缺必填字段，其余行完好（观察逐行结果与计数）
+            field = _first_required_field(entity)
+            if bulk_rows:
+                if len(bulk_rows) < 2:
+                    # 样例仅 1 行时复制一行（新幂等键），保证「部分失败」可演示
+                    extra = json.loads(json.dumps(bulk_rows[0], ensure_ascii=False))
+                    extra["idempotencyKey"] = f"{run_id}-L{len(bulk_rows) + 1}"
+                    bulk_rows.append(extra)
+                bulk_rows[-1]["data"]["list"][0].pop(field, None)
         status, body, headers = client.op_bulk(entity, bulk_rows, contract_version=contract_version)
         layers = _layers("SUPPORTED", config_state, "ACCEPTED" if status == 202 else "RUN_FAILED")
         return record(status, body, headers, layers)
@@ -871,6 +981,234 @@ def runs():
     return {"runs": RUNS.recent()}
 
 
+# ---------------------------------------------------------------------------
+# 异常测试（第二阶段 N1-N4，任务书 §12-§15）
+# ---------------------------------------------------------------------------
+
+@app.get("/api/fault/catalog")
+def fault_catalog_route():
+    return {"scenarios": fault_catalog.FAULT_SCENARIOS,
+            "counts": fault_catalog.scenario_counts(),
+            "reportFields": fault_catalog.FAULT_REPORT_FIELDS}
+
+
+def _apply_db_fault(directive: dict, object_code: str) -> dict:
+    """执行 N1 模拟库异常预置（只操作模拟库，硬阻断保证不触碰 Ai-DOP 库）。"""
+    obj = _require_obj(object_code)
+    db_spec = obj.get("dbSync") or {}
+    table = db_spec.get("table")
+    if not table:
+        raise HTTPException(status_code=400, detail=f"{object_code} 无 dbSync 表（N1 场景需要）")
+    if not _env("AIDOP_SIM_MYSQL_PASSWORD"):
+        raise HTTPException(status_code=400,
+                            detail="缺少 AIDOP_SIM_MYSQL_PASSWORD 环境变量：模拟库操作需要 MySQL 凭据")
+    rows = sa.adapt_db_rows(obj, sa.new_run_id())
+    spec = {"rows": rows, "businessKey": db_spec.get("businessKey"),
+            "incrementColumn": db_spec.get("incrementColumn", "sourceUpdatedAt")}
+    action = directive["type"]
+    try:
+        if action == "db_drop_table":
+            mock_db.drop_table(table)
+            return {"ok": True, "table": table, "note": f"已删除源表 {table}（Ai-DOP 侧保持配置不变再触发同步）"}
+        if action == "db_load_null_bizkey":
+            mock_db.drop_table(table)   # 从正常结构出发，保证场景独立可复现
+            mock_db.init_schema({table: spec})
+            null_row = {k: (None if k == "bizKey" else v) for k, v in rows[0].items()}
+            mock_db.insert_rows_raw(table, [null_row])
+            return {"ok": True, "table": table, "note": f"已向 {table} 追加一行 bizKey=NULL 的行"}
+        if action == "db_load_duplicate_bizkey":
+            # 必须先 DROP：init_schema 的 DDL 是 CREATE TABLE IF NOT EXISTS，
+            # 表已存在时不会重建，旧唯一键会残留导致重复行插入报 1062。
+            mock_db.drop_table(table)
+            mock_db.init_schema({table: {**spec, "unique_key": False}})
+            mock_db.insert_rows_raw(table, [rows[0], dict(rows[0])])
+            return {"ok": True, "table": table,
+                    "note": f"{table} 已重建为无唯一键表并装载两条相同 bizKey 的行"}
+        if action == "db_update_no_advance":
+            mock_db.drop_table(table)   # 从正常结构出发，保证场景独立可复现
+            mock_db.init_schema({table: spec})
+            mock_db.upsert_rows(table, rows)
+            mock_db.update_rows(table, {"itemName": "N1-08 更新后不推进增量列"},
+                                {"bizKey": rows[0]["bizKey"]}, spec["incrementColumn"],
+                                advance_increment=False)
+            return {"ok": True, "table": table, "note": f"已更新 {table} 一行但未推进增量列"}
+        if action == "db_slow_view":
+            mock_db.init_schema({table: spec})
+            view = f"v_slow_{table}"
+            mock_db.create_slow_view(view, table, float(directive.get("sleepSeconds", 0.5)))
+            return {"ok": True, "view": view, "table": table,
+                    "note": f"已创建慢查询视图 {view}；请测试人员在 Ai-DOP 侧把 source_table_name 改为该视图名"}
+    except HTTPException:
+        raise
+    except Exception as ex:
+        raise HTTPException(status_code=502, detail=f"{type(ex).__name__}: {ex}")
+
+
+def _apply_mock_api_fault(directive: dict) -> dict:
+    s = MOCK_STATE.settings
+    settings = directive.get("settings") or {}
+    if settings.get("reset"):
+        s.reset_faults()
+    for key in ("force_status", "delay_seconds", "timeout_seconds", "empty_response",
+                "invalid_json", "ignore_cursor", "no_sort"):
+        if key in settings:
+            setattr(s, key, settings[key])
+    for key in ("content_type", "response_path"):
+        if key in settings:
+            setattr(s, key, settings[key])
+    return {"ok": True, "settings": mock_api_settings(),
+            "mockRunning": _mock_running(),
+            "note": "" if _mock_running() else "提示：Mock API 未启动，先点「启动 Mock API」"}
+
+
+def _apply_mock_api_override(directive: dict, object_code: str) -> dict:
+    obj = _require_obj(object_code)
+    path = (obj.get("apiPull") or {}).get("path") or "/api/item"
+    rows = sa.adapt_api_rows(obj, sa.new_run_id())
+    mode = directive.get("mode")
+    if mode == "missing_bizkey":
+        for r in rows:
+            r.pop("bizKey", None)
+    elif mode == "duplicate_bizkey":
+        rows = rows + [dict(rows[0])]
+    MOCK_STATE.overrides[path] = rows
+    return {"ok": True, "overridePath": path, "rowCount": len(rows),
+            "mode": mode,
+            "note": "已挂载覆盖行；Ai-DOP 拉取该 path 时生效（恢复默认时清除）"}
+
+
+@app.post("/api/fault/apply")
+def fault_apply(payload: FaultApplyPayload):
+    scenario = fault_catalog.find_scenario(payload.scenarioId)
+    if not scenario:
+        raise HTTPException(status_code=404, detail=f"unknown scenarioId: {payload.scenarioId}")
+    if scenario["kind"] == "guide":
+        return {"applied": False, "guide": True, "scenario": scenario,
+                "note": "此场景需测试人员在 Ai-DOP 侧人工操作（模拟器不修改 Ai-DOP 配置）"}
+    directive = scenario["apply"]
+    dtype = directive["type"]
+    if dtype.startswith("db_"):
+        return {"applied": True, "scenario": scenario,
+                ** _apply_db_fault(directive, payload.objectCode or "ITEM")}
+    if dtype == "mock_api":
+        return {"applied": True, "scenario": scenario,
+                **_apply_mock_api_fault(directive)}
+    if dtype == "mock_api_override":
+        return {"applied": True, "scenario": scenario,
+                **_apply_mock_api_override(directive, payload.objectCode or "ITEM")}
+    if dtype == "inbound_case":
+        case = directive["case"]
+        object_code = payload.objectCode or "ITEM"
+        if case == "disabled-entity":
+            entity = (payload.entityCode or "S5_INVENTORY_TXN").upper()
+            try:
+                pre = precheck_inbound_one(entity)
+            except HTTPException as ex:
+                pre = {"httpStatus": ex.status_code, "configState": "UNREACHABLE",
+                       "message": ex.detail}
+            return {"applied": True, "scenario": scenario, "entityCode": entity,
+                    "precheck": pre,
+                    "note": "请测试人员先在 Ai-DOP 侧关闭该实体 inbound 开关，再用「API_INBOUND」页签向它推送；"
+                            "被 Ai-DOP 正确拒绝 = 测试通过"}
+        if case == "snapshot-commit-failure":
+            exec_payload = InboundExecPayload(objectCode=object_code,
+                                              operation="snapshot_commit",
+                                              snapshotId="SNAP-NOT-EXIST-0001")
+        elif case == "bulk-partial-failure":
+            exec_payload = InboundExecPayload(objectCode=object_code, operation="bulk",
+                                              case="bulk-partial-failure")
+        elif case == "unknown-entity":
+            exec_payload = InboundExecPayload(objectCode=object_code, operation="push",
+                                              case="unknown-entity")
+        else:
+            exec_payload = InboundExecPayload(objectCode=object_code, operation="push", case=case)
+        try:
+            result = inbound_exec(exec_payload)
+        except HTTPException as ex:
+            # 仅「Ai-DOP 预检拦截」（409 + dict detail）才属于异常场景的「实际响应」：
+            # 正确拒绝 = 测试通过。模拟器自身的配置缺失（缺环境变量/白名单拦截等 400）
+            # 必须如实透传，不得包装成 expectedReject，否则会把「工具未发出请求」
+            # 误报成「Ai-DOP 拒绝了请求」。
+            if ex.status_code == 409 and isinstance(ex.detail, dict):
+                return {"applied": True, "scenario": scenario, "expectedReject": True,
+                        "httpStatus": ex.status_code, "detail": ex.detail,
+                        "note": "Ai-DOP 已在预检阶段拒绝（正确拒绝属于测试通过，不属于工具失败）"}
+            raise
+        return {"applied": True, "scenario": scenario, "result": result}
+    raise HTTPException(status_code=400, detail=f"unknown apply directive: {dtype}")
+
+
+@app.post("/api/fault/recover")
+def fault_recover(payload: dict):
+    """恢复默认：channel 可传 DB_SYNC|API_PULL|API_INBOUND，缺省全部。"""
+    channel = payload.get("channel")
+    object_code = payload.get("objectCode") or "ITEM"
+    notes: list[str] = []
+    if channel in (None, "API_PULL"):
+        MOCK_STATE.settings.reset_faults()
+        cleared = len(MOCK_STATE.overrides)
+        MOCK_STATE.overrides.clear()
+        notes.append(f"Mock API 异常注入已全部恢复默认；清除 {cleared} 个覆盖行挂载"
+                     f"（场景编排挂载需重跑场景恢复）")
+    if channel in (None, "DB_SYNC"):
+        obj = sa.find_object(object_code)
+        db_spec = (obj or {}).get("dbSync") or {}
+        table = db_spec.get("table")
+        if not table:
+            notes.append(f"{object_code} 无 dbSync 表，跳过模拟库恢复")
+        elif not _env("AIDOP_SIM_MYSQL_PASSWORD"):
+            notes.append("缺少 AIDOP_SIM_MYSQL_PASSWORD：模拟库恢复未执行"
+                         "（设置环境变量后重试；或按场景说明人工恢复）")
+        else:
+            try:
+                rows = sa.adapt_db_rows(obj, sa.new_run_id())
+                spec = {"rows": rows, "businessKey": db_spec.get("businessKey"),
+                        "incrementColumn": db_spec.get("incrementColumn", "sourceUpdatedAt")}
+                # 先 DROP 再建：init_schema 的 DDL 是 CREATE TABLE IF NOT EXISTS，
+                # 不先删则 N1-07 的无唯一键表结构会残留，恢复后仍是重复键表。
+                mock_db.drop_table(table)
+                mock_db.init_schema({table: spec})   # 重建带唯一键的正常表
+                mock_db.upsert_rows(table, rows)
+                mock_db.drop_view(f"v_slow_{table}")
+                notes.append(f"{table} 已重建为正常表并装载样例行；慢查询视图已删除")
+            except Exception as ex:
+                notes.append(f"模拟库恢复失败：{type(ex).__name__}: {ex}")
+    if channel in (None, "API_INBOUND"):
+        notes.append("方式丙无状态注入无需恢复；Ai-DOP 侧实体开关由测试人员自行恢复")
+    return {"ok": True, "notes": notes}
+
+
+@app.post("/api/fault-report/save")
+def fault_report_save(payload: FaultReportPayload):
+    if not fault_catalog.find_scenario(payload.scenarioId):
+        raise HTTPException(status_code=400, detail=f"unknown scenarioId: {payload.scenarioId}")
+    entry = FAULT_REPORTS.save(payload.model_dump(by_alias=True), _secrets(), _access_keys())
+    return {"ok": True, "report": entry}
+
+
+@app.get("/api/fault-report/list")
+def fault_report_list():
+    return {"reports": FAULT_REPORTS.list()}
+
+
+@app.get("/api/fault-report/export/{report_id}")
+def fault_report_export(report_id: str):
+    entry = FAULT_REPORTS.get(report_id)
+    if not entry:
+        raise HTTPException(status_code=404, detail="report not found")
+    md = build_fault_report_markdown(entry)
+    filename = f"aidop-sim-fault-report-{entry['scenarioId']}-{report_id}.md"
+    return Response(content=md, media_type="text/markdown; charset=utf-8",
+                    headers={"Content-Disposition": f'attachment; filename="{filename}"'})
+
+
+@app.delete("/api/fault-report/{report_id}")
+def fault_report_delete(report_id: str):
+    if not FAULT_REPORTS.delete(report_id):
+        raise HTTPException(status_code=404, detail="report not found")
+    return {"ok": True}
+
+
 # ---------------------------------------------------------------------------
 # 全链路报告（F5）
 # ---------------------------------------------------------------------------

tools/integration-simulator/clients/inbound_client.py

@@ -68,11 +68,15 @@ class InboundClient:
         contract_version: str | None = None,
         extra_headers: dict | None = None,
         sign_path: str | None = None,
+        skip_access_key: bool = False,
+        access_key_override: str | None = None,
     ) -> tuple[int, dict | str, dict]:
         """返回 (http_status, parsed_body, sent_headers)。sent_headers 中签名已遮蔽。
 
         sign_path：签名串路径与请求 URL 路径不同时使用（查询串不得进入签名，
-        对齐 Ai-DOP InboundSignatureHandler 用 Request.Path.Value 验签）。"""
+        对齐 Ai-DOP InboundSignatureHandler 用 Request.Path.Value 验签）。
+        skip_access_key：不发送 X-Access-Key（N3 缺 AccessKey）。
+        access_key_override：用指定值替代 AccessKey（N3 错误 AccessKey，签名自洽但未知）。"""
         raw = body if body is not None else b""
         digest = sha256_hex(raw)
         if tamper_body:
@@ -80,18 +84,20 @@ class InboundClient:
         ts_s = str(ts if ts is not None else int(time.time()))
         nonce = nonce or uuid.uuid4().hex
         idem = idem or f"IDEM-{uuid.uuid4().hex}"
-        message = build_message(method, sign_path or path, self.key, ts_s, nonce, digest, idem)
+        sent_key = access_key_override or self.key
+        message = build_message(method, sign_path or path, sent_key, ts_s, nonce, digest, idem)
         signature = sign(self._secret, message)
         if bad_signature:
             signature = sign(self._secret + "-wrong", message)
         headers = {
             "Content-Type": "application/json",
-            "X-Access-Key": self.key,
             "X-Timestamp": ts_s,
             "X-Nonce": nonce,
             "X-Content-SHA256": digest,
             "X-Signature": signature,
         }
+        if not skip_access_key:
+            headers["X-Access-Key"] = sent_key
         if not skip_idem:
             headers["Idempotency-Key"] = idem
         if contract_version:
@@ -188,3 +194,51 @@ class InboundClient:
 
     def case_bad_signature(self, entity: str, envelope: dict):
         return self.op_push(entity, envelope, bad_signature=True)
+
+    # -- 第二阶段 N3 新增用例 ------------------------------------------------
+
+    def case_missing_access_key(self, entity: str, envelope: dict):
+        """缺 AccessKey：不发送 X-Access-Key 头。"""
+        return self.op_push(entity, envelope, skip_access_key=True)
+
+    def case_wrong_access_key(self, entity: str, envelope: dict):
+        """错误 AccessKey：签名自洽但 AccessKey 未授权。"""
+        return self.op_push(entity, envelope, access_key_override="SIM-UNKNOWN-ACCESS-KEY")
+
+    def case_missing_required_field(self, entity: str, envelope: dict, field: str):
+        """缺必填字段：逐行删除指定契约字段。"""
+        env = json.loads(json.dumps(envelope, ensure_ascii=False))
+        for row in env["data"]["list"]:
+            row.pop(field, None)
+        return self.op_push(entity, env)
+
+    def case_wrong_field_type(self, entity: str, envelope: dict, field: str, bad_value):
+        """字段类型错误：把指定字段替换为错误类型值（如数值列塞字符串）。"""
+        env = json.loads(json.dumps(envelope, ensure_ascii=False))
+        for row in env["data"]["list"]:
+            row[field] = bad_value
+        return self.op_push(entity, env)
+
+    def case_stale_rows(self, entity: str, envelope: dict):
+        """stale 数据：逐行把时间类字段退到 2000 年（快照 commit 时被 Ai-DOP 拒绝）。"""
+        env = json.loads(json.dumps(envelope, ensure_ascii=False))
+        for row in env["data"]["list"]:
+            for key in tuple(row):
+                if str(key).lower() in ("sourceupdatedat", "updatedat", "bizdate", "occurredat"):
+                    row[key] = "2000-01-01T00:00:00"
+        return self.op_push(entity, env)
+
+    def case_unknown_entity(self, envelope: dict):
+        """未授权/未注册 entityCode：推送到一个不存在于目录的实体码。"""
+        return self.op_push("SIM_UNKNOWN_ENTITY_XX", envelope)
+
+    def case_bulk_partial_failure(self, entity: str, ndjson_lines: list[dict], field: str,
+                                  bad_idx: int = -1):
+        """bulk 部分失败：指定行删除必填字段，其余行完好。"""
+        lines = [json.loads(json.dumps(line, ensure_ascii=False)) for line in ndjson_lines]
+        lines[bad_idx].pop(field, None)
+        return self.op_bulk(entity, lines)
+
+    def case_snapshot_commit_failure(self, entity: str):
+        """snapshot commit 失败：提交一个不存在的 snapshotId。"""
+        return self.op_snapshot_commit(entity, "SNAP-NOT-EXIST-0001")

tools/integration-simulator/providers/fault_catalog.py

@@ -0,0 +1,244 @@
+"""第二阶段异常场景目录（任务书 §12-§15）。
+
+纯数据 + 应用/恢复指令：副作用由 app.py 的 fault_apply / fault_recover 执行，
+本模块不连接数据库、不改任何状态，保证场景定义可单测、可审计。
+
+kind 语义：
+- auto  模拟器一键预置（apply 指令由模拟器执行）
+- guide 需测试人员在 Ai-DOP 侧人工操作（页面展示指引，模拟器只控制第三方一侧）
+
+被 Ai-DOP 正确拒绝属于测试通过，不属于工具失败。
+"""
+from __future__ import annotations
+
+# 场景报告八列（任务书 §15）
+FAULT_REPORT_FIELDS: list[tuple[str, str]] = [
+    ("scenarioId", "场景"),
+    ("simBehavior", "模拟端行为"),
+    ("expected", "预期结果"),
+    ("actualResponse", "实际响应"),
+    ("aidopFlowStatus", "Ai-DOP 流水状态"),
+    ("targetChanged", "目标数据是否变化"),
+    ("pass", "是否符合预期"),
+    ("note", "备注"),
+]
+
+# ---------------------------------------------------------------------------
+# N1 方式甲 DB_SYNC（10 场景）
+# ---------------------------------------------------------------------------
+
+N1_SCENARIOS: list[dict] = [
+    {"id": "N1-01", "channel": "DB_SYNC", "kind": "guide", "title": "数据库地址/端口错误",
+     "sim": "测试人员在 Ai-DOP 侧把来源连接串改为未监听端口（如 3307）。模拟器不修改 Ai-DOP 配置。",
+     "expect": "Ai-DOP 测连或同步失败，提示连接被拒绝/超时。",
+     "recover": "把 Ai-DOP 侧连接串改回正确地址。"},
+    {"id": "N1-02", "channel": "DB_SYNC", "kind": "guide", "title": "数据库用户名/密码错误",
+     "sim": "测试人员在 Ai-DOP 侧填写错误的测试账号/密码（如 sim_user/wrong）。",
+     "expect": "认证失败（Access denied），同步任务失败。",
+     "recover": "改回正确测试账号。"},
+    {"id": "N1-03", "channel": "DB_SYNC", "kind": "guide", "title": "无表权限",
+     "sim": "测试人员在模拟库上创建受限账号（仅 SELECT 部分表或完全无权限），Ai-DOP 侧配置该账号。",
+     "expect": "查询被拒绝（SELECT command denied）。",
+     "recover": "Ai-DOP 侧换回有权限账号。"},
+    {"id": "N1-04", "channel": "DB_SYNC", "kind": "auto", "title": "源表不存在",
+     "sim": "模拟器删除源表（DROP TABLE）。测试人员在 Ai-DOP 侧保持配置不变并触发同步。",
+     "expect": "明确报表不存在（Table ... doesn't exist），同步失败不静默。",
+     "recover": "点击恢复：按目录样例重建表并装载样例行。",
+     "apply": {"type": "db_drop_table"}},
+    {"id": "N1-05", "channel": "DB_SYNC", "kind": "guide", "title": "字段不存在（错误增量列）",
+     "sim": "测试人员在 Ai-DOP 侧把增量列配置改为不存在的列名（如 sourceUpdatedAt2）。",
+     "expect": "明确报列不存在（Unknown column），同步失败。",
+     "recover": "改回正确增量列名。"},
+    {"id": "N1-06", "channel": "DB_SYNC", "kind": "auto", "title": "空业务键",
+     "sim": "模拟器装载 bizKey 为 NULL 的行（额外插入，不清空原数据）。",
+     "expect": "被拒绝或隔离，不静默覆盖正常数据。",
+     "recover": "点击恢复：删除空键行并重建样例行。",
+     "apply": {"type": "db_load_null_bizkey"}},
+    {"id": "N1-07", "channel": "DB_SYNC", "kind": "auto", "title": "重复业务键",
+     "sim": "模拟器重建无唯一键的表并装载两条相同 bizKey 的行。",
+     "expect": "行为可解释（Ai-DOP 侧去重/告警），不产生脏数据。",
+     "recover": "点击恢复：重建带唯一键的表并装载样例行。",
+     "apply": {"type": "db_load_duplicate_bizkey"}},
+    {"id": "N1-08", "channel": "DB_SYNC", "kind": "auto", "title": "增量列不推进",
+     "sim": "模拟器更新一行数据但保持增量列时间不变（advance_increment=false）。",
+     "expect": "该行不被误判为新增量；下一次正常推进后可同步。",
+     "recover": "点击恢复：重建样例行。",
+     "apply": {"type": "db_update_no_advance"}},
+    {"id": "N1-09", "channel": "DB_SYNC", "kind": "auto", "title": "慢查询",
+     "sim": "模拟器创建受控延迟视图 v_slow_{table}（每行 SLEEP 0.5 秒），并提示测试人员在 Ai-DOP 侧把 source_table_name 改为该视图名。",
+     "expect": "Ai-DOP 读取耗时显著变长，超时行为可解释。",
+     "recover": "Ai-DOP 侧改回原表名；点击恢复删除视图。",
+     "apply": {"type": "db_slow_view", "sleepSeconds": 0.5}},
+    {"id": "N1-10", "channel": "DB_SYNC", "kind": "guide", "title": "连接中断",
+     "sim": "测试人员在同步过程中停止模拟库服务（或断开网络），Ai-DOP 侧触发同步。",
+     "expect": "失败不记成功，重连后状态可恢复。",
+     "recover": "重新启动模拟库服务。"},
+]
+
+# ---------------------------------------------------------------------------
+# N2 方式乙 API_PULL（16 场景）
+# ---------------------------------------------------------------------------
+
+N2_SCENARIOS: list[dict] = [
+    {"id": "N2-01", "channel": "API_PULL", "kind": "auto", "title": "401 未授权",
+     "sim": "Mock API 返回 401（force_status=401）。",
+     "expect": "Ai-DOP 拉取失败，流水记录失败状态，不写入数据。",
+     "recover": "恢复默认。", "apply": {"type": "mock_api", "settings": {"force_status": 401}}},
+    {"id": "N2-02", "channel": "API_PULL", "kind": "auto", "title": "403 禁止",
+     "sim": "Mock API 返回 403（force_status=403）。",
+     "expect": "Ai-DOP 拉取失败，流水记录失败状态。",
+     "recover": "恢复默认。", "apply": {"type": "mock_api", "settings": {"force_status": 403}}},
+    {"id": "N2-03", "channel": "API_PULL", "kind": "auto", "title": "404 路径不存在",
+     "sim": "Mock API 返回 404（force_status=404）。",
+     "expect": "Ai-DOP 拉取失败，明确提示路径不存在。",
+     "recover": "恢复默认。", "apply": {"type": "mock_api", "settings": {"force_status": 404}}},
+    {"id": "N2-04", "channel": "API_PULL", "kind": "auto", "title": "429 限流",
+     "sim": "Mock API 返回 429（force_status=429）。",
+     "expect": "Ai-DOP 拉取失败或按限流策略重试。",
+     "recover": "恢复默认。", "apply": {"type": "mock_api", "settings": {"force_status": 429}}},
+    {"id": "N2-05", "channel": "API_PULL", "kind": "auto", "title": "500 服务器错误",
+     "sim": "Mock API 返回 500（force_status=500）。",
+     "expect": "Ai-DOP 拉取失败，错误信息明确。",
+     "recover": "恢复默认。", "apply": {"type": "mock_api", "settings": {"force_status": 500}}},
+    {"id": "N2-06", "channel": "API_PULL", "kind": "auto", "title": "503 服务不可用",
+     "sim": "Mock API 返回 503（force_status=503）。",
+     "expect": "Ai-DOP 拉取失败，错误信息明确。",
+     "recover": "恢复默认。", "apply": {"type": "mock_api", "settings": {"force_status": 503}}},
+    {"id": "N2-07", "channel": "API_PULL", "kind": "auto", "title": "固定延迟",
+     "sim": "Mock API 每次响应前延迟 3 秒（delay_seconds=3）。",
+     "expect": "Ai-DOP 响应耗时约 3 秒，请求成功。",
+     "recover": "恢复默认。", "apply": {"type": "mock_api", "settings": {"delay_seconds": 3}}},
+    {"id": "N2-08", "channel": "API_PULL", "kind": "auto", "title": "超时",
+     "sim": "Mock API 睡眠 120 秒（timeout_seconds=120），超过 Ai-DOP 读取超时。",
+     "expect": "Ai-DOP 侧超时终止，流水记录失败。",
+     "recover": "恢复默认。", "apply": {"type": "mock_api", "settings": {"timeout_seconds": 120}}},
+    {"id": "N2-09", "channel": "API_PULL", "kind": "auto", "title": "空响应",
+     "sim": "Mock API 返回 200 空 body（empty_response=true）。",
+     "expect": "Ai-DOP 解析失败并明确报错，不写入数据。",
+     "recover": "恢复默认。", "apply": {"type": "mock_api", "settings": {"empty_response": True}}},
+    {"id": "N2-10", "channel": "API_PULL", "kind": "auto", "title": "非法 JSON",
+     "sim": "Mock API 返回截断的非法 JSON 文本（invalid_json=true）。",
+     "expect": "Ai-DOP 解析失败并明确报错。",
+     "recover": "恢复默认。", "apply": {"type": "mock_api", "settings": {"invalid_json": True}}},
+    {"id": "N2-11", "channel": "API_PULL", "kind": "auto", "title": "错误 Content-Type",
+     "sim": "Mock API 返回 text/html 的 Content-Type（content_type=text/html）。",
+     "expect": "Ai-DOP 按 Content-Type 校验失败或仍按 JSON 解析，行为可解释。",
+     "recover": "恢复默认。", "apply": {"type": "mock_api", "settings": {"content_type": "text/html"}}},
+    {"id": "N2-12", "channel": "API_PULL", "kind": "auto", "title": "错误 response path",
+     "sim": "Mock API 把数据挂到不存在的路径 data.notExist（response_path=data.notExist）。",
+     "expect": "Ai-DOP 取不到列表，明确报错。",
+     "recover": "恢复默认。", "apply": {"type": "mock_api", "settings": {"response_path": "data.notExist"}}},
+    {"id": "N2-13", "channel": "API_PULL", "kind": "auto", "title": "缺 bizKey",
+     "sim": "Mock API 挂载去掉 bizKey 字段的覆盖行（/api/item）。",
+     "expect": "Ai-DOP 去重键缺失，报错或隔离，不静默写入。",
+     "recover": "恢复默认（清除覆盖行）。", "apply": {"type": "mock_api_override", "mode": "missing_bizkey"}},
+    {"id": "N2-14", "channel": "API_PULL", "kind": "auto", "title": "重复 bizKey",
+     "sim": "Mock API 挂载两条相同 bizKey 的覆盖行（/api/item）。",
+     "expect": "Ai-DOP 去重行为可解释，不产生脏数据。",
+     "recover": "恢复默认（清除覆盖行）。", "apply": {"type": "mock_api_override", "mode": "duplicate_bizkey"}},
+    {"id": "N2-15", "channel": "API_PULL", "kind": "auto", "title": "cursor 不推进",
+     "sim": "Mock API 无视 cursor 返回全部行（ignore_cursor=true）。",
+     "expect": "Ai-DOP 每次拿到同样数据，去重后无新增，行为可解释。",
+     "recover": "恢复默认。", "apply": {"type": "mock_api", "settings": {"ignore_cursor": True}}},
+    {"id": "N2-16", "channel": "API_PULL", "kind": "auto", "title": "cursor 乱序",
+     "sim": "Mock API 不按 bizKey 排序返回（no_sort=true）。",
+     "expect": "Ai-DOP 分页去重行为可解释，不产生脏数据。",
+     "recover": "恢复默认。", "apply": {"type": "mock_api", "settings": {"no_sort": True}}},
+]
+
+# ---------------------------------------------------------------------------
+# N3 方式丙 API_INBOUND（17 场景，任务书 §14 原文）
+# ---------------------------------------------------------------------------
+
+N3_SCENARIOS: list[dict] = [
+    {"id": "N3-01", "channel": "API_INBOUND", "kind": "auto", "title": "缺 AccessKey",
+     "sim": "推数请求不发送 X-Access-Key 头。",
+     "expect": "Ai-DOP 拒绝（401/403 授权缺失），不产生流水数据。",
+     "recover": "无需恢复（无状态注入）。", "apply": {"type": "inbound_case", "case": "missing-access-key"}},
+    {"id": "N3-02", "channel": "API_INBOUND", "kind": "auto", "title": "错误 AccessKey",
+     "sim": "发送未授权的 AccessKey（SIM-UNKNOWN-ACCESS-KEY，签名自洽）。",
+     "expect": "Ai-DOP 拒绝（授权缺失/未知 AccessKey）。",
+     "recover": "无需恢复。", "apply": {"type": "inbound_case", "case": "wrong-access-key"}},
+    {"id": "N3-03", "channel": "API_INBOUND", "kind": "auto", "title": "错误 Secret/签名",
+     "sim": "用错误 Secret 计算签名（bad_signature=true）。",
+     "expect": "Ai-DOP 验签失败（401/403）。",
+     "recover": "无需恢复。", "apply": {"type": "inbound_case", "case": "bad-signature"}},
+    {"id": "N3-04", "channel": "API_INBOUND", "kind": "auto", "title": "body 摘要不一致",
+     "sim": "发送后篡改 body（X-Content-SHA256 与实际不符）。",
+     "expect": "Ai-DOP 摘要校验失败，明确拒绝。",
+     "recover": "无需恢复。", "apply": {"type": "inbound_case", "case": "tamper-body"}},
+    {"id": "N3-05", "channel": "API_INBOUND", "kind": "auto", "title": "缺幂等键",
+     "sim": "请求不发送 Idempotency-Key 头。",
+     "expect": "Ai-DOP 拒绝（幂等键必填）。",
+     "recover": "无需恢复。", "apply": {"type": "inbound_case", "case": "missing-idem"}},
+    {"id": "N3-06", "channel": "API_INBOUND", "kind": "auto", "title": "过期 timestamp",
+     "sim": "X-Timestamp 设为 1 小时前。",
+     "expect": "Ai-DOP 拒绝（时间戳超出允许窗口）。",
+     "recover": "无需恢复。", "apply": {"type": "inbound_case", "case": "expired-ts"}},
+    {"id": "N3-07", "channel": "API_INBOUND", "kind": "auto", "title": "nonce replay",
+     "sim": "相同 nonce 连发两个请求。",
+     "expect": "第二个被拒绝（防重放）。",
+     "recover": "无需恢复。", "apply": {"type": "inbound_case", "case": "nonce-replay"}},
+    {"id": "N3-08", "channel": "API_INBOUND", "kind": "auto", "title": "同幂等键同 body",
+     "sim": "相同 Idempotency-Key 与 body 连发两次。",
+     "expect": "第二次命中幂等返回相同结果，不重复落库。",
+     "recover": "无需恢复。", "apply": {"type": "inbound_case", "case": "same-key-replay"}},
+    {"id": "N3-09", "channel": "API_INBOUND", "kind": "auto", "title": "同幂等键不同 body",
+     "sim": "相同 Idempotency-Key 但 body 不同。",
+     "expect": "Ai-DOP 拒绝（幂等键冲突）。",
+     "recover": "无需恢复。", "apply": {"type": "inbound_case", "case": "same-key-diff-body"}},
+    {"id": "N3-10", "channel": "API_INBOUND", "kind": "auto", "title": "缺必填字段",
+     "sim": "推数行删除契约必填字段（如 MDM_ITEM 的 Descr）。",
+     "expect": "Ai-DOP 拒绝或隔离该行（映射/校验失败）。",
+     "recover": "无需恢复。", "apply": {"type": "inbound_case", "case": "missing-required-field"}},
+    {"id": "N3-11", "channel": "API_INBOUND", "kind": "auto", "title": "字段类型错误",
+     "sim": "推数行把必填字段替换为结构敌对值（嵌套对象，实测比数字更能逼近类型校验边界）。",
+     "expect": "Ai-DOP 拒绝或隔离该行（类型校验失败）；若宽容转换则在报告中记录实证。",
+     "recover": "无需恢复。", "apply": {"type": "inbound_case", "case": "wrong-field-type"}},
+    {"id": "N3-12", "channel": "API_INBOUND", "kind": "auto", "title": "部分行拒绝",
+     "sim": "push 信封多行中一行缺必填字段。",
+     "expect": "坏行被拒绝，好行正常入站（或整体拒绝），行为可解释。",
+     "recover": "无需恢复。", "apply": {"type": "inbound_case", "case": "partial-reject"}},
+    {"id": "N3-13", "channel": "API_INBOUND", "kind": "auto", "title": "stale 数据",
+     "sim": "推数行时间字段退到 2000 年（快照场景下模拟旧数据）。",
+     "expect": "Ai-DOP 拒绝过期数据或按 stale 策略处理。",
+     "recover": "无需恢复。", "apply": {"type": "inbound_case", "case": "stale"}},
+    {"id": "N3-14", "channel": "API_INBOUND", "kind": "auto", "title": "未授权 entityCode",
+     "sim": "推送到不存在/未授权的实体码 SIM_UNKNOWN_ENTITY_XX。",
+     "expect": "Ai-DOP 返回 404 实体未注册（或 403 未授权）。",
+     "recover": "无需恢复。", "apply": {"type": "inbound_case", "case": "unknown-entity"}},
+    {"id": "N3-15", "channel": "API_INBOUND", "kind": "auto", "title": "未启用 entity",
+     "sim": "测试人员先在 Ai-DOP 侧关闭某实体 inbound 开关，模拟器再向该实体推送（默认 S5_INVENTORY_TXN，可填其他已注册实体码）。",
+     "expect": "Ai-DOP 拒绝（实体未启用）。",
+     "recover": "Ai-DOP 侧重新启用该实体。",
+     "apply": {"type": "inbound_case", "case": "disabled-entity"}},
+    {"id": "N3-16", "channel": "API_INBOUND", "kind": "auto", "title": "bulk 部分失败",
+     "sim": "bulk 端点 ndjson 多行中一行缺必填字段。",
+     "expect": "Ai-DOP 返回逐行结果，坏行计数准确。",
+     "recover": "无需恢复。", "apply": {"type": "inbound_case", "case": "bulk-partial-failure"}},
+    {"id": "N3-17", "channel": "API_INBOUND", "kind": "auto", "title": "snapshot commit 失败",
+     "sim": "提交不存在的 snapshotId（SNAP-NOT-EXIST-0001）。",
+     "expect": "Ai-DOP 拒绝 commit，明确报错。",
+     "recover": "无需恢复。", "apply": {"type": "inbound_case", "case": "snapshot-commit-failure"}},
+]
+
+FAULT_SCENARIOS: list[dict] = N1_SCENARIOS + N2_SCENARIOS + N3_SCENARIOS
+
+FAULT_CHANNEL_LABELS = {
+    "DB_SYNC": "方式甲 DB_SYNC（数据库同步）",
+    "API_PULL": "方式乙 API_PULL（接口拉取）",
+    "API_INBOUND": "方式丙 API_INBOUND（接口推送）",
+}
+
+
+def scenario_counts() -> dict:
+    counts = {"DB_SYNC": 0, "API_PULL": 0, "API_INBOUND": 0, "total": 0}
+    for s in FAULT_SCENARIOS:
+        counts[s["channel"]] += 1
+        counts["total"] += 1
+    return counts
+
+
+def find_scenario(scenario_id: str) -> dict | None:
+    sid = scenario_id.upper()
+    return next((s for s in FAULT_SCENARIOS if s["id"].upper() == sid), None)

tools/integration-simulator/providers/mock_api.py

@@ -23,7 +23,7 @@ _SIM_ROOT = Path(__file__).resolve().parents[1]
 if str(_SIM_ROOT) not in sys.path:
     sys.path.insert(0, str(_SIM_ROOT))
 
-from fastapi import FastAPI, HTTPException, Query, Request  # noqa: E402
+from fastapi import FastAPI, HTTPException, Query, Request, Response  # noqa: E402
 
 from catalog.sample_adapters import MOCK_API_DIR  # noqa: E402
 from clients.redaction import redact_headers  # noqa: E402
@@ -38,9 +38,29 @@ class MockApiSettings:
     api_key_header: str = "X-Api-Key"
     api_key_value: str = "sim-api-key"
     delay_seconds: float = 0.0
-    force_status: int | None = None   # 401/403/429/500 等
+    force_status: int | None = None   # 401/403/404/429/500/503 等
     response_path: str = "data.list"
     audit_limit: int = 200
+    # ---- 第二阶段 N2 异常模式（内存态，恢复默认即清除）----
+    timeout_seconds: float = 0.0      # >0 时按秒数睡眠，模拟超过 Ai-DOP 超时
+    empty_response: bool = False      # 返回 200 空 body
+    invalid_json: bool = False        # 返回非法 JSON 文本
+    content_type: str | None = None   # 覆盖响应 Content-Type（如 text/html）
+    ignore_cursor: bool = False       # 无视 cursor 返回全部行（cursor 不推进）
+    no_sort: bool = False             # 不按 bizKey 排序（cursor 乱序）
+
+    def reset_faults(self) -> None:
+        """恢复默认：清空全部异常注入设置（鉴权/凭据类保留）。"""
+        self.auth_mode = "TOKEN"
+        self.delay_seconds = 0.0
+        self.force_status = None
+        self.response_path = "data.list"
+        self.timeout_seconds = 0.0
+        self.empty_response = False
+        self.invalid_json = False
+        self.content_type = None
+        self.ignore_cursor = False
+        self.no_sort = False
 
 
 @dataclass
@@ -100,6 +120,24 @@ def _check_auth(state: MockApiState, request: Request) -> None:
         raise HTTPException(status_code=401, detail=f"unsupported auth mode on mock side: {mode}")
 
 
+def _active_faults(s: MockApiSettings) -> list[str]:
+    """当前生效的异常注入清单（写入审计，便于人工核对「选择的异常」）。"""
+    faults: list[str] = []
+    if s.timeout_seconds > 0:
+        faults.append(f"timeout:{s.timeout_seconds}s")
+    if s.empty_response:
+        faults.append("emptyResponse")
+    if s.invalid_json:
+        faults.append("invalidJson")
+    if s.content_type:
+        faults.append(f"contentType:{s.content_type}")
+    if s.ignore_cursor:
+        faults.append("ignoreCursor")
+    if s.no_sort:
+        faults.append("noSort")
+    return faults
+
+
 def _audit(state: MockApiState, entry: dict) -> None:
     state.audit.append(entry)
     if len(state.audit) > state.settings.audit_limit:
@@ -138,11 +176,17 @@ def create_app(state: MockApiState | None = None) -> FastAPI:
             "authMode": s.auth_mode, "delaySeconds": s.delay_seconds,
             "forceStatus": s.force_status, "responsePath": s.response_path,
             "apiKeyHeader": s.api_key_header,
+            "timeoutSeconds": s.timeout_seconds, "emptyResponse": s.empty_response,
+            "invalidJson": s.invalid_json, "contentType": s.content_type,
+            "ignoreCursor": s.ignore_cursor, "noSort": s.no_sort,
         }
 
     @app.post("/__control/settings")
     def update_settings(payload: dict):
         s = state.settings
+        if payload.get("reset"):
+            s.reset_faults()
+            return get_settings()
         if "auth_mode" in payload:
             mode = str(payload["auth_mode"]).upper()
             if mode not in ("NONE", "TOKEN", "BASIC", "APIKEY", "OAUTH2"):
@@ -164,6 +208,18 @@ def create_app(state: MockApiState | None = None) -> FastAPI:
             s.force_status = int(payload["force_status"]) if payload["force_status"] else None
         if "response_path" in payload and payload["response_path"]:
             s.response_path = str(payload["response_path"])
+        if "timeout_seconds" in payload:
+            s.timeout_seconds = max(0.0, float(payload["timeout_seconds"]))
+        if "empty_response" in payload:
+            s.empty_response = bool(payload["empty_response"])
+        if "invalid_json" in payload:
+            s.invalid_json = bool(payload["invalid_json"])
+        if "content_type" in payload:
+            s.content_type = str(payload["content_type"]) if payload["content_type"] else None
+        if "ignore_cursor" in payload:
+            s.ignore_cursor = bool(payload["ignore_cursor"])
+        if "no_sort" in payload:
+            s.no_sort = bool(payload["no_sort"])
         return get_settings()
 
     @app.post("/__control/overrides")
@@ -192,12 +248,21 @@ def create_app(state: MockApiState | None = None) -> FastAPI:
     ):
         started = time.time()
         path = request.url.path
-        if state.settings.delay_seconds > 0:
-            time.sleep(state.settings.delay_seconds)
+        s = state.settings
         try:
-            if state.settings.force_status:
-                raise HTTPException(status_code=state.settings.force_status,
-                                    detail=f"forced error {state.settings.force_status}")
+            if s.timeout_seconds > 0:
+                # 超时模拟：睡眠超过 Ai-DOP 读取超时（由测试人员配置秒数）
+                time.sleep(s.timeout_seconds)
+            if s.empty_response:
+                return Response(content="", media_type=s.content_type or "application/json")
+            if s.invalid_json:
+                return Response(content='{"data": {"list": [',
+                                media_type=s.content_type or "application/json")
+            if s.delay_seconds > 0:
+                time.sleep(s.delay_seconds)
+            if s.force_status:
+                raise HTTPException(status_code=s.force_status,
+                                    detail=f"forced error {s.force_status}")
             _check_auth(state, request)
             if path in state.overrides:
                 rows = [dict(r) for r in state.overrides[path]]
@@ -207,11 +272,15 @@ def create_app(state: MockApiState | None = None) -> FastAPI:
                     raise HTTPException(status_code=404,
                                         detail=f"no mock for {path}; add it to endpoints.json or set an override")
                 rows = _load_rows(endpoints[path])
-            rows.sort(key=lambda r: str(r.get("bizKey", "")))
-            total = len(rows)
-            if cursor:
+            if not s.no_sort:
+                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]
-            return _build_response(rows, state.settings.response_path)
+            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
         finally:
             _audit(state, {
                 "time": time.strftime("%Y-%m-%dT%H:%M:%S%z"),
@@ -219,6 +288,7 @@ def create_app(state: MockApiState | None = None) -> FastAPI:
                 "cursor": cursor,
                 "authMode": state.settings.auth_mode,
                 "forceStatus": state.settings.force_status,
+                "faults": _active_faults(state.settings),
                 "elapsedMs": int((time.time() - started) * 1000),
                 "headers": redact_headers(dict(request.headers), state.secrets()),
             })

tools/integration-simulator/providers/mock_db.py

@@ -85,8 +85,10 @@ def infer_mysql_type(values: list) -> str:
 
 
 def generate_ddl(table: str, rows: list[dict], business_key: list[str] | None = None,
-                 increment_column: str = "sourceUpdatedAt") -> str:
-    """按样例行推断列类型；业务键唯一索引；增量列建普通索引。列名保持样例原样（反引号包裹）。"""
+                 increment_column: str = "sourceUpdatedAt", unique_key: bool = True) -> str:
+    """按样例行推断列类型；业务键唯一索引；增量列建普通索引。列名保持样例原样（反引号包裹）。
+
+    unique_key=False：不建业务键唯一索引（N1 重复业务键场景）。"""
     if not rows:
         raise ValueError(f"cannot generate DDL for {table} from empty rows")
     business_key = business_key or ["bizKey"]
@@ -97,9 +99,10 @@ def generate_ddl(table: str, rows: list[dict], business_key: list[str] | None =
     lines = []
     for name, values in columns.items():
         lines.append(f"  `{name}` {infer_mysql_type(values)} NULL")
-    uk = ", ".join(f"`{k}`" for k in business_key if k in columns)
-    if uk:
-        lines.append(f"  UNIQUE KEY `uk_{table}` ({uk})")
+    if unique_key:
+        uk = ", ".join(f"`{k}`" for k in business_key if k in columns)
+        if uk:
+            lines.append(f"  UNIQUE KEY `uk_{table}` ({uk})")
     if increment_column in columns:
         lines.append(f"  KEY `ix_{table}_incr` (`{increment_column}`)")
     return (f"CREATE TABLE IF NOT EXISTS `{table}` (\n" + ",\n".join(lines) +
@@ -148,7 +151,8 @@ def init_schema(tables: dict[str, dict]) -> dict:
             result["database"] = database
             for table, spec in tables.items():
                 ddl = generate_ddl(table, spec["rows"], spec.get("businessKey"),
-                                   spec.get("incrementColumn", "sourceUpdatedAt"))
+                                   spec.get("incrementColumn", "sourceUpdatedAt"),
+                                   unique_key=spec.get("unique_key", True))
                 cur.execute(ddl)
                 result["tables"].append(table)
     finally:
@@ -187,10 +191,13 @@ def upsert_rows(table: str, rows: list[dict]) -> int:
 
 def update_rows(table: str, updates: dict, where: dict,
                 increment_column: str = "sourceUpdatedAt",
-                new_increment: str | None = None) -> int:
-    """更新已有数据并推进增量列（DB_SYNC 增量更新用例）。"""
+                new_increment: str | None = None,
+                advance_increment: bool = True) -> int:
+    """更新已有数据并推进增量列（DB_SYNC 增量更新用例）。
+
+    advance_increment=False：不推进增量列（N1 增量列不推进场景）。"""
     sets = {**updates}
-    if increment_column:
+    if increment_column and advance_increment:
         sets[increment_column] = new_increment or datetime.now().strftime("%Y-%m-%dT%H:%M:%S")
     set_clause = ", ".join(f"`{k}`=%s" for k in sets)
     where_clause = " AND ".join(f"`{k}`=%s" for k in where)
@@ -204,6 +211,60 @@ def update_rows(table: str, updates: dict, where: dict,
         conn.close()
 
 
+def insert_rows_raw(table: str, rows: list[dict]) -> int:
+    """原始 INSERT（不合并）：供 N1 重复业务键场景装载（表须无唯一键）。"""
+    if not rows:
+        return 0
+    cols = _col_names(rows)
+    names = ", ".join(f"`{c}`" for c in cols)
+    marks = ", ".join(["%s"] * len(cols))
+    sql = f"INSERT INTO `{table}` ({names}) VALUES ({marks})"
+    conn = _use_db()
+    try:
+        with conn.cursor() as cur:
+            cur.executemany(sql, [tuple(_convert(row.get(c)) for c in cols) for row in rows])
+            return len(rows)
+    finally:
+        conn.close()
+
+
+def drop_table(table: str) -> None:
+    """删除源表（N1 源表不存在场景；恢复时用 ensure 重建）。"""
+    conn = _use_db()
+    try:
+        with conn.cursor() as cur:
+            cur.execute(f"DROP TABLE IF EXISTS `{table}`")
+    finally:
+        conn.close()
+
+
+def build_slow_view_ddl(view: str, table: str, sleep_seconds: float) -> str:
+    """受控延迟视图 DDL（纯函数可单测）：每行 SLEEP 固定秒数。
+
+    Ai-DOP 侧把实体 source_table_name 人工配置为该视图名即可制造慢查询。
+    """
+    return (f"CREATE OR REPLACE VIEW `{view}` AS "
+            f"SELECT t.*, SLEEP({sleep_seconds}) AS `__sim_sleep` FROM `{table}` t")
+
+
+def create_slow_view(view: str, table: str, sleep_seconds: float) -> None:
+    conn = _use_db()
+    try:
+        with conn.cursor() as cur:
+            cur.execute(build_slow_view_ddl(view, table, sleep_seconds))
+    finally:
+        conn.close()
+
+
+def drop_view(view: str) -> None:
+    conn = _use_db()
+    try:
+        with conn.cursor() as cur:
+            cur.execute(f"DROP VIEW IF EXISTS `{view}`")
+    finally:
+        conn.close()
+
+
 def table_stats(table: str, increment_column: str = "sourceUpdatedAt") -> dict:
     conn = _use_db()
     try:

tools/integration-simulator/static/app.css

@@ -169,3 +169,19 @@ button:disabled { opacity: .5; cursor: not-allowed; }
 .arrow { color: var(--muted); }
 
 code { background: #eef2f7; padding: 1px 4px; border-radius: 3px; }
+
+/* ---- Òì³£²âÊÔҳǩ£¨µÚ¶þ½×¶Î N1-N4£© ---- */
+.fault-item {
+  border: 1px solid var(--line);
+  border-radius: 6px;
+  padding: 8px 10px;
+  margin: 6px 0;
+  background: #f8fafc;
+  font-size: 12px;
+}
+.fault-item .muted { margin: 2px 0; }
+.fault-head { display: flex; align-items: center; gap: 8px; margin-bottom: 4px; font-size: 13px; }
+.fault-head b { color: var(--accent, #3730a3); }
+.chip.ok { background: #f0fdf4; border-color: #bbf7d0; color: #166534; }
+.fault-result.ok { color: #166534; }
+.fault-result.err { color: var(--err); }

tools/integration-simulator/static/app.js

@@ -80,6 +80,9 @@ async function refreshStatus() {
     $('mock-state').textContent = s.mockApi.running
       ? `运行中 :${s.mockApi.port}（${s.mockApi.authMode}）` : '未运行';
     $('mock-state').className = 'state ' + (s.mockApi.running ? 'ok' : 'warn');
+    // 异常注入状态同步回 Mock 页下拉框（fault 页预置 N2 场景后这里应显示对应状态码）
+    const force = s.mockApi.forceStatus;
+    $('mock-force').value = force ? String(force) : '';
     show('security-output', s);
     const counts = s.objectCounts;
     $('run-object-info').dataset.counts = JSON.stringify(counts);
@@ -270,6 +273,152 @@ async function inboundExec() {
   catch (ex) { setError('inbound-output', ex); }
 }
 
+/* ---------------- 异常测试（第二阶段 N1-N4） ---------------- */
+let FAULT_CATALOG = null;
+let LAST_FAULT_REPORT_ID = null;
+
+const FAULT_GROUP_TITLES = {
+  DB_SYNC: '方式甲 DB_SYNC（数据库同步）· 10 场景',
+  API_PULL: '方式乙 API_PULL（接口拉取）· 16 场景',
+  API_INBOUND: '方式丙 API_INBOUND（接口推送）· 17 场景',
+};
+
+function faultScenarios() { return (FAULT_CATALOG && FAULT_CATALOG.scenarios) || []; }
+
+function esc(s) {
+  return String(s).replace(/&/g, '&amp;').replace(/</g, '&lt;').replace(/>/g, '&gt;');
+}
+
+async function refreshFaultCatalog() {
+  try {
+    FAULT_CATALOG = await api('/api/fault/catalog');
+    renderFaultGroups();
+    fillFaultReportScenario();
+  } catch (ex) { setError('fault-groups', ex); }
+}
+
+function renderFaultGroups() {
+  const container = $('fault-groups');
+  const byChannel = { DB_SYNC: [], API_PULL: [], API_INBOUND: [] };
+  faultScenarios().forEach((s) => (byChannel[s.channel] || []).push(s));
+  container.innerHTML = ['DB_SYNC', 'API_PULL', 'API_INBOUND'].map((ch) => {
+    const items = (byChannel[ch] || []).map((s) => `
+      <div class="fault-item" data-id="${s.id}">
+        <div class="fault-head">
+          <b>${s.id}</b> ${esc(s.title)}
+          <span class="chip ${s.kind === 'auto' ? 'ok' : ''}">${s.kind === 'auto' ? 'auto · 一键预置' : 'guide · 人工操作'}</span>
+        </div>
+        <div class="muted">模拟方法：${esc(s.sim)}</div>
+        <div class="muted">预期观察：${esc(s.expect)}</div>
+        <div class="muted">恢复：${esc(s.recover)}</div>
+        <div class="row">
+          <button data-act="apply">${s.kind === 'auto' ? '预置此异常' : '查看指引'}</button>
+          <button data-act="recover">恢复</button>
+          <span class="fault-result muted"></span>
+        </div>
+      </div>`).join('');
+    return `<h3>${FAULT_GROUP_TITLES[ch]}</h3>${items}`;
+  }).join('');
+
+  container.querySelectorAll('.fault-item').forEach((item) => {
+    const id = item.dataset.id;
+    item.querySelector('button[data-act="apply"]').addEventListener('click', async () => {
+      const out = item.querySelector('.fault-result');
+      out.textContent = '预置中…';
+      try {
+        const r = await api('/api/fault/apply', { method: 'POST', body: JSON.stringify({ scenarioId: id }) });
+        const note = r.note || (r.result ? JSON.stringify(r.result).slice(0, 120) : '');
+        out.textContent = '✓ ' + (r.applied ? '已预置' : '请按指引操作') + (note ? ' · ' + note : '');
+        out.className = 'fault-result ok';
+      } catch (ex) {
+        out.textContent = '❌ ' + ex.message;
+        out.className = 'fault-result err';
+      }
+    });
+    item.querySelector('button[data-act="recover"]').addEventListener('click', async () => {
+      const out = item.querySelector('.fault-result');
+      const scenario = faultScenarios().find((s) => s.id === id);
+      out.textContent = '恢复中…';
+      try {
+        const r = await api('/api/fault/recover', {
+          method: 'POST', body: JSON.stringify({ channel: scenario ? scenario.channel : undefined }),
+        });
+        out.textContent = '✓ 已恢复：' + (r.notes || []).join('；');
+        out.className = 'fault-result ok';
+      } catch (ex) {
+        out.textContent = '❌ ' + ex.message;
+        out.className = 'fault-result err';
+      }
+    });
+  });
+}
+
+function fillFaultReportScenario() {
+  const sel = $('fault-report-scenario');
+  sel.innerHTML = '';
+  faultScenarios().forEach((s) => {
+    const opt = document.createElement('option');
+    opt.value = s.id;
+    opt.textContent = `${s.id} · ${s.title}`;
+    sel.appendChild(opt);
+  });
+  syncFaultReportPrefill();
+}
+
+function syncFaultReportPrefill() {
+  const s = faultScenarios().find((x) => x.id === $('fault-report-scenario').value);
+  if (!s) return;
+  $('fault-report-sim').value = s.sim;
+  $('fault-report-expect').value = s.expect;
+}
+
+async function refreshFaultReportList() {
+  try { show('fault-report-list', await api('/api/fault-report/list')); }
+  catch (ex) { setError('fault-report-list', ex); }
+}
+
+async function saveFaultReport() {
+  const btn = $('btn-fault-report-save');
+  if (btn) btn.disabled = true;  // 防重复点击（同场景二次保存也会被后端原位更新）
+  const fields = {
+    scenarioId: $('fault-report-scenario').value,
+    simBehavior: $('fault-report-sim').value.trim(),
+    expected: $('fault-report-expect').value.trim(),
+    actualResponse: $('fault-report-actual').value.trim(),
+    aidopFlowStatus: $('fault-report-flow').value.trim(),
+    targetChanged: $('fault-report-target').value.trim(),
+    pass: $('fault-report-pass').value,
+    note: $('fault-report-note').value.trim(),
+  };
+  try {
+    const r = await api('/api/fault-report/save', { method: 'POST', body: JSON.stringify(fields) });
+    LAST_FAULT_REPORT_ID = r.report.id;
+    await refreshFaultReportList();
+    alert(r.report.replaced
+      ? `已更新场景 ${r.report.scenarioId} 的报告（同场景覆盖，不产生重复行）`
+      : `已保存异常报告 ${r.report.id}（保存前已脱敏）`);
+  } catch (ex) { setError('fault-report-list', ex); }
+  finally { if (btn) btn.disabled = false; }
+}
+
+function exportFaultReport(id) {
+  if (!id) { alert('请先「保存报告」，再导出 Markdown'); return; }
+  const a = document.createElement('a');
+  a.href = '/api/fault-report/export/' + encodeURIComponent(id);
+  a.download = `aidop-sim-fault-report-${id}.md`;
+  document.body.appendChild(a);
+  a.click();
+  a.remove();
+}
+
+async function recoverAllFaults() {
+  try {
+    const r = await api('/api/fault/recover', { method: 'POST', body: JSON.stringify({}) });
+    alert('已恢复：\n' + (r.notes || []).join('\n'));
+    refreshStatus();
+  } catch (ex) { alert('恢复失败：' + ex.message); }
+}
+
 /* ---------------- 全链路报告（F5） ---------------- */
 let LAST_REPORT_ID = null;
 
@@ -316,7 +465,7 @@ function exportReport(id) {
   if (!id) { alert('请先「保存报告」，再导出 Markdown'); return; }
   const a = document.createElement('a');
   a.href = '/api/report/export/' + encodeURIComponent(id);
-  a.download = '';
+  a.download = `aidop-sim-report-${id}.md`;
   document.body.appendChild(a);
   a.click();
   a.remove();
@@ -411,6 +560,14 @@ async function init() {
   $('btn-report-evidence').addEventListener('click', importReportEvidence);
   refreshReportList();
 
+  $('btn-fault-refresh').addEventListener('click', refreshFaultCatalog);
+  $('btn-fault-recover-all').addEventListener('click', recoverAllFaults);
+  $('fault-report-scenario').addEventListener('change', syncFaultReportPrefill);
+  $('btn-fault-report-save').addEventListener('click', saveFaultReport);
+  $('btn-fault-report-export').addEventListener('click', () => exportFaultReport(LAST_FAULT_REPORT_ID));
+  $('btn-fault-report-refresh').addEventListener('click', refreshFaultReportList);
+  refreshFaultCatalog();
+
   refreshChannelState();
   renderScenarioGraph();
   await refreshStatus();

tools/integration-simulator/static/index.html

@@ -18,6 +18,7 @@
     <button data-tab="tab-mockapi">Mock API（API_PULL）</button>
     <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-report">全链路报告</button>
     <button data-tab="tab-runs">运行记录</button>
     <button data-tab="tab-security">环境安全</button>
@@ -106,8 +107,10 @@
             <option value="">不强制</option>
             <option value="401">401</option>
             <option value="403">403</option>
+            <option value="404">404</option>
             <option value="429">429</option>
             <option value="500">500</option>
+            <option value="503">503</option>
           </select>
         </label>
         <label>响应路径 <input id="mock-resp-path" type="text" value="data.list" /></label>
@@ -193,6 +196,52 @@
     </div>
   </section>
 
+  <!-- 异常测试（第二阶段 N1-N4） -->
+  <section id="tab-fault" class="tab">
+    <div class="card">
+      <h2>异常测试场景（第二阶段 · N1 甲 10 / N2 乙 16 / N3 丙 17）</h2>
+      <p class="muted">
+        auto = 模拟器一键预置；guide = 按说明在 Ai-DOP 侧人工操作（模拟器不修改 Ai-DOP 配置）。
+        被 Ai-DOP 正确拒绝 = 测试通过，不属于工具失败。
+      </p>
+      <div class="row">
+        <button id="btn-fault-refresh">刷新目录</button>
+        <button id="btn-fault-recover-all" class="primary">一键恢复全部默认</button>
+        <span class="muted">异常均为内存态注入，重启控制台亦自动恢复</span>
+      </div>
+      <div id="fault-groups"></div>
+    </div>
+    <div class="grid2">
+      <div class="card">
+        <h2>异常测试报告（N4 · 八列）</h2>
+        <p class="muted">保存前自动脱敏；选择场景后自动带入模拟端行为与预期结果。</p>
+        <label>场景 <select id="fault-report-scenario"></select></label>
+        <label>模拟端行为 <textarea id="fault-report-sim" rows="2"></textarea></label>
+        <label>预期结果 <textarea id="fault-report-expect" rows="2"></textarea></label>
+        <label>实际响应 <textarea id="fault-report-actual" rows="2" placeholder="Ai-DOP 实际响应/报错（人工填写）"></textarea></label>
+        <label>Ai-DOP 流水状态 <input id="fault-report-flow" type="text" placeholder="mdp_sync_log / mdp_inbound_request 状态" /></label>
+        <label>目标数据是否变化 <input id="fault-report-target" type="text" placeholder="如：贴源表无新增行" /></label>
+        <label>是否符合预期
+          <select id="fault-report-pass">
+            <option value="是">是</option>
+            <option value="否">否</option>
+            <option value="部分">部分</option>
+          </select>
+        </label>
+        <label>备注 <textarea id="fault-report-note" rows="2"></textarea></label>
+        <div class="row">
+          <button id="btn-fault-report-save" class="primary">保存报告（自动脱敏）</button>
+          <button id="btn-fault-report-export">导出最新 Markdown</button>
+          <button id="btn-fault-report-refresh">刷新列表</button>
+        </div>
+      </div>
+      <div class="card">
+        <h2>已保存异常报告（内存态，重启即清空）</h2>
+        <pre id="fault-report-list" class="output small">—</pre>
+      </div>
+    </div>
+  </section>
+
   <!-- 全链路报告 -->
   <section id="tab-report" class="tab">
     <div class="grid2">

tools/integration-simulator/tests/test_fault.py

@@ -0,0 +1,649 @@
+"""第二阶段（N1-N4）异常测试能力守卫。
+
+- N2：Mock API 异常注入模式（超时/空响应/非法 JSON/Content-Type/cursor 行为/恢复默认）
+- N1：模拟库异常能力（drop 表/无唯一键重复装载/增量列不推进/慢查询视图 DDL）
+- N3：入站客户端异常用例（缺/错 AccessKey、缺字段、类型错误、stale、snapshot 失败）
+- N4：异常测试报告（字段覆盖任务书 §15 八列、脱敏、导出）
+
+全部进程内验证（TestClient/纯函数），不依赖 Ai-DOP 在线与真实 MySQL。
+"""
+from __future__ import annotations
+
+import json
+
+import pytest
+
+fastapi_testclient = pytest.importorskip("fastapi.testclient")
+TestClient = fastapi_testclient.TestClient
+
+from providers import mock_api  # noqa: E402
+from providers import mock_db  # noqa: E402
+from catalog import sample_adapters as sa  # noqa: E402
+
+
+# ---------------------------------------------------------------------------
+# N2：Mock API 异常模式
+# ---------------------------------------------------------------------------
+
+@pytest.fixture()
+def client():
+    state = mock_api.MockApiState()
+    state.settings.auth_mode = "NONE"
+    return TestClient(mock_api.create_app(state)), state
+
+
+def _set(client, **kw):
+    c, state = client
+    for key, val in kw.items():
+        setattr(state.settings, key, val)
+
+
+def test_force_status_covers_n2_codes(client):
+    """N2 要求 401/403/404/429/500/503：force_status 须对全部六类生效。"""
+    c, _ = client
+    for code in (401, 403, 404, 429, 500, 503):
+        _set(client, force_status=code)
+        assert c.get("/api/item").status_code == code, code
+    _set(client, force_status=None)
+    assert c.get("/api/item").status_code == 200
+
+
+def test_empty_response_mode(client):
+    c, _ = client
+    _set(client, empty_response=True)
+    resp = c.get("/api/item")
+    assert resp.status_code == 200
+    assert resp.content == b""
+
+
+def test_invalid_json_mode(client):
+    c, _ = client
+    _set(client, invalid_json=True)
+    resp = c.get("/api/item")
+    assert resp.status_code == 200
+    with pytest.raises(Exception):
+        resp.json()  # 无法解析为 JSON
+
+
+def test_wrong_content_type_mode(client):
+    c, _ = client
+    _set(client, content_type="text/html")
+    resp = c.get("/api/item")
+    assert resp.status_code == 200
+    assert resp.headers["content-type"].startswith("text/html")
+
+
+def test_timeout_mode_sleeps_configured_seconds(client):
+    c, _ = client
+    _set(client, timeout_seconds=0.3)
+    import time
+    started = time.time()
+    resp = c.get("/api/item")
+    assert resp.status_code == 200
+    assert time.time() - started >= 0.3
+
+
+def test_ignore_cursor_returns_all_rows(client):
+    c, _ = client
+    rows = [
+        {"bizKey": "K-1", "v": 1}, {"bizKey": "K-2", "v": 2}, {"bizKey": "K-3", "v": 3},
+    ]
+    c.post("/__control/overrides", json={"path": "/api/order-test", "rows": rows})
+    # 默认 cursor 过滤：只返回更大行
+    after = c.get("/api/order-test", params={"cursor": "K-1"}).json()["data"]["list"]
+    assert [r["bizKey"] for r in after] == ["K-2", "K-3"]
+    # ignore_cursor：无视 cursor 返回全部行（cursor 不推进观察用）
+    _set(client, ignore_cursor=True)
+    after = c.get("/api/order-test", params={"cursor": "K-1"}).json()["data"]["list"]
+    assert [r["bizKey"] for r in after] == ["K-1", "K-2", "K-3"]
+
+
+def test_no_sort_keeps_original_order(client):
+    c, state = client
+    rows = [
+        {"bizKey": "Z-2"}, {"bizKey": "A-1"}, {"bizKey": "M-3"},
+    ]
+    c.post("/__control/overrides", json={"path": "/api/order-test", "rows": rows})
+    # 默认排序
+    got = [r["bizKey"] for r in c.get("/api/order-test").json()["data"]["list"]]
+    assert got == ["A-1", "M-3", "Z-2"]
+    # no_sort 保持挂载原顺序（cursor 乱序观察用）
+    _set(client, no_sort=True)
+    got = [r["bizKey"] for r in c.get("/api/order-test").json()["data"]["list"]]
+    assert got == ["Z-2", "A-1", "M-3"]
+
+
+def test_reset_faults_clears_all_injections(client):
+    c, _ = client
+    _set(client, force_status=500, empty_response=True, invalid_json=True,
+         content_type="text/html", ignore_cursor=True, no_sort=True,
+         timeout_seconds=1.0, delay_seconds=0.5)
+    resp = c.post("/__control/settings", json={"reset": True}).json()
+    assert resp["forceStatus"] is None
+    assert resp["emptyResponse"] is False
+    assert resp["invalidJson"] is False
+    assert resp["contentType"] is None
+    assert resp["ignoreCursor"] is False
+    assert resp["noSort"] is False
+    assert resp["timeoutSeconds"] == 0.0
+    assert resp["delaySeconds"] == 0.0
+    assert resp["authMode"] == "TOKEN"  # 鉴权模式一并恢复默认
+    # 恢复后正常拉取可用
+    _set(client, auth_mode="NONE")
+    assert c.get("/api/item").status_code == 200
+
+
+def test_audit_records_active_faults(client):
+    c, _ = client
+    _set(client, empty_response=True, ignore_cursor=True)
+    c.get("/api/item")
+    entries = c.get("/__control/audit").json()["entries"]
+    assert entries, "audit should record the request"
+    faults = entries[-1]["faults"]
+    assert "emptyResponse" in faults
+    assert "ignoreCursor" in faults
+
+
+# ---------------------------------------------------------------------------
+# N1：模拟库异常能力（纯函数层；真实 MySQL 用例沿用 test_mock_db.py 的跳过约定）
+# ---------------------------------------------------------------------------
+
+RUN_ID = "20260930120000-abc123"
+
+
+def test_ddl_without_unique_key_for_duplicate_bizkey():
+    rows = sa.adapt_db_rows(sa.find_object("ITEM"), RUN_ID)
+    ddl = mock_db.generate_ddl("sim_item", rows, ["bizKey"], "sourceUpdatedAt",
+                               unique_key=False)
+    assert "UNIQUE KEY" not in ddl  # 重复业务键场景：无唯一索引，允许重复行共存
+    assert "CREATE TABLE IF NOT EXISTS `sim_item`" in ddl
+
+
+def test_raw_insert_sql_has_no_duplicate_merge():
+    """N1 重复业务键：原始 INSERT 不合并（与 upsert 的 ON DUPLICATE KEY 不同）。"""
+
+    class FakeCur:
+        def __init__(self):
+            self.executed = []
+
+        def executemany(self, sql, rows):
+            self.executed.append(sql)
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *args):
+            return False
+
+    class FakeConn:
+        def __init__(self, cur):
+            self._cur = cur
+
+        def cursor(self):
+            return self._cur
+
+        def close(self):
+            pass
+
+    cur = FakeCur()
+    monkeypatch = pytest.MonkeyPatch()
+    monkeypatch.setattr(mock_db, "_use_db", lambda: FakeConn(cur))
+    try:
+        mock_db.insert_rows_raw("sim_item", [{"bizKey": "K-1", "v": 1},
+                                             {"bizKey": "K-1", "v": 2}])
+    finally:
+        monkeypatch.undo()
+    assert cur.executed and "ON DUPLICATE KEY" not in cur.executed[0]
+
+
+def test_update_rows_can_skip_increment_advance():
+    """N1 增量列不推进：advance_increment=False 时 SET 子句不含增量列。"""
+
+    class FakeCur:
+        def __init__(self):
+            self.executed = []
+
+        def execute(self, sql, args=None):
+            self.executed.append(sql)
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *args):
+            return False
+
+        @property
+        def rowcount(self):
+            return 1
+
+    class FakeConn:
+        def __init__(self, cur):
+            self._cur = cur
+
+        def cursor(self):
+            return self._cur
+
+        def close(self):
+            pass
+
+    cur = FakeCur()
+    monkeypatch = pytest.MonkeyPatch()
+    monkeypatch.setattr(mock_db, "_use_db", lambda: FakeConn(cur))
+    try:
+        mock_db.update_rows("sim_item", {"itemName": "改后"}, {"bizKey": "K-1"},
+                            "sourceUpdatedAt", advance_increment=False)
+    finally:
+        monkeypatch.undo()
+    assert cur.executed
+    sql = cur.executed[0]
+    assert "`itemName`=%s" in sql
+    assert "sourceUpdatedAt" not in sql  # 增量列未被写入
+
+
+def test_slow_view_ddl_is_controllable():
+    ddl = mock_db.build_slow_view_ddl("v_slow_sim_item", "sim_item", 0.5)
+    assert ddl.startswith("CREATE OR REPLACE VIEW `v_slow_sim_item`")
+    assert "SLEEP(0.5)" in ddl
+    assert "FROM `sim_item` t" in ddl
+
+
+def test_drop_sql_targets_simulator_table():
+    """N1 源表不存在：drop 只在模拟库执行（表名反引号包裹，无库名拼接）。"""
+
+    class FakeCur:
+        def __init__(self):
+            self.executed = []
+
+        def execute(self, sql):
+            self.executed.append(sql)
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *args):
+            return False
+
+    class FakeConn:
+        def __init__(self, cur):
+            self._cur = cur
+
+        def cursor(self):
+            return self._cur
+
+        def close(self):
+            pass
+
+    cur = FakeCur()
+    monkeypatch = pytest.MonkeyPatch()
+    monkeypatch.setattr(mock_db, "_use_db", lambda: FakeConn(cur))
+    try:
+        mock_db.drop_table("sim_item")
+        mock_db.drop_view("v_slow_sim_item")
+    finally:
+        monkeypatch.undo()
+    assert cur.executed == ["DROP TABLE IF EXISTS `sim_item`",
+                            "DROP VIEW IF EXISTS `v_slow_sim_item`"]
+
+
+# ---------------------------------------------------------------------------
+# N3：入站客户端异常用例（本地捕获请求验证，不触达 Ai-DOP）
+# ---------------------------------------------------------------------------
+
+from clients import inbound_client as ic  # noqa: E402
+
+
+def _fake_urlopen(monkeypatch, captured: list):
+    class FakeResp:
+        status = 200
+
+        def read(self):
+            return b'{"code": 0}'
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *args):
+            return False
+
+    def fake(req, timeout=None):
+        captured.append({
+            "url": req.full_url,
+            "headers": dict(req.headers),
+            "data": req.data,
+        })
+        return FakeResp()
+
+    monkeypatch.setattr(ic.urllib.request, "urlopen", fake)
+    return captured
+
+
+@pytest.fixture()
+def envelope():
+    return {"data": {"list": [
+        {"ItemNum": "SIM-1", "Descr": "甲", "sourceUpdatedAt": "2026-10-01T00:00:00"},
+        {"ItemNum": "SIM-2", "Descr": "乙", "sourceUpdatedAt": "2026-10-02T00:00:00"},
+    ]}}
+
+
+def _client():
+    return ic.InboundClient("http://127.0.0.1:5005", "SIM-ACCESS-KEY", "sim-secret")
+
+
+def _h(headers: dict, name: str):
+    """urllib 会把 header 名 title-case 化（X-access-key），大小写不敏感取值。"""
+    for k, v in headers.items():
+        if k.lower() == name.lower():
+            return v
+    raise KeyError(name)
+
+
+def test_case_missing_access_key_omits_header(monkeypatch, envelope):
+    cap = []
+    _fake_urlopen(monkeypatch, cap)
+    _client().case_missing_access_key("MDM_ITEM", envelope)
+    assert not any(k.lower() == "x-access-key" for k in cap[0]["headers"])
+
+
+def test_case_wrong_access_key_is_self_consistent(monkeypatch, envelope):
+    cap = []
+    _fake_urlopen(monkeypatch, cap)
+    _client().case_wrong_access_key("MDM_ITEM", envelope)
+    sent = cap[0]
+    assert _h(sent["headers"], "X-Access-Key") == "SIM-UNKNOWN-ACCESS-KEY"
+    # 签名自洽：用发送的 key 重算应等于请求头中的签名
+    ts = _h(sent["headers"], "X-Timestamp")
+    nonce = _h(sent["headers"], "X-Nonce")
+    digest = _h(sent["headers"], "X-Content-SHA256")
+    idem = _h(sent["headers"], "Idempotency-Key")
+    msg = ic.build_message("POST", "/api/mdp/inbound/MDM_ITEM",
+                           "SIM-UNKNOWN-ACCESS-KEY", ts, nonce, digest, idem)
+    assert _h(sent["headers"], "X-Signature") == ic.sign("sim-secret", msg)
+
+
+def test_case_missing_required_field_drops_field(monkeypatch, envelope):
+    cap = []
+    _fake_urlopen(monkeypatch, cap)
+    _client().case_missing_required_field("MDM_ITEM", envelope, "Descr")
+    body = json.loads(cap[0]["data"].decode("utf-8"))
+    for row in body["data"]["list"]:
+        assert "Descr" not in row
+    assert envelope["data"]["list"][0]["Descr"] == "甲"  # 原始信封未被修改
+
+
+def test_case_wrong_field_type_replaces_value(monkeypatch, envelope):
+    cap = []
+    _fake_urlopen(monkeypatch, cap)
+    _client().case_wrong_field_type("MDM_ITEM", envelope, "Descr", 12345)
+    body = json.loads(cap[0]["data"].decode("utf-8"))
+    assert body["data"]["list"][0]["Descr"] == 12345
+
+
+def test_case_stale_rows_backdates_timestamp(monkeypatch, envelope):
+    cap = []
+    _fake_urlopen(monkeypatch, cap)
+    _client().case_stale_rows("MDM_ITEM", envelope)
+    body = json.loads(cap[0]["data"].decode("utf-8"))
+    for row in body["data"]["list"]:
+        assert row["sourceUpdatedAt"] == "2000-01-01T00:00:00"
+
+
+def test_case_unknown_entity_targets_unknown_code(monkeypatch, envelope):
+    cap = []
+    _fake_urlopen(monkeypatch, cap)
+    _client().case_unknown_entity(envelope)
+    assert cap[0]["url"].endswith("/api/mdp/inbound/SIM_UNKNOWN_ENTITY_XX")
+
+
+def test_case_bulk_partial_failure_drops_one_line_field(monkeypatch, envelope):
+    cap = []
+    _fake_urlopen(monkeypatch, cap)
+    lines = envelope["data"]["list"]
+    _client().case_bulk_partial_failure("MDM_ITEM", lines, "Descr", bad_idx=1)
+    assert _h(cap[0]["headers"], "Content-Type") == "application/x-ndjson"
+    ndjson = cap[0]["data"].decode("utf-8").splitlines()
+    row0, row1 = json.loads(ndjson[0]), json.loads(ndjson[1])
+    assert "Descr" in row0 and "Descr" not in row1  # 仅坏行缺字段
+
+
+def test_case_snapshot_commit_failure_targets_missing_snapshot(monkeypatch):
+    cap = []
+    _fake_urlopen(monkeypatch, cap)
+    _client().case_snapshot_commit_failure("MDM_ITEM")
+    assert "SNAP-NOT-EXIST-0001" in cap[0]["url"]
+
+
+# ---------------------------------------------------------------------------
+# N4：场景目录、预置/恢复、异常报告（app.py 路由，TestClient 进程内）
+# ---------------------------------------------------------------------------
+
+import app  # noqa: E402  conftest 已把模拟器根目录加入 sys.path
+
+
+@pytest.fixture()
+def app_client():
+    return TestClient(app.app)
+
+
+def test_fault_catalog_matches_taskbook_counts(app_client):
+    """任务书 §12-§14：甲 10 / 乙 16 / 丙 17（原文清单数），N4 八列。"""
+    body = app_client.get("/api/fault/catalog").json()
+    counts = body["counts"]
+    assert counts == {"DB_SYNC": 10, "API_PULL": 16, "API_INBOUND": 17, "total": 43}
+    labels = [label for _, label in body["reportFields"]]
+    assert labels == ["场景", "模拟端行为", "预期结果", "实际响应",
+                      "Ai-DOP 流水状态", "目标数据是否变化", "是否符合预期", "备注"]
+    # 每个场景都必须带恢复说明（验收：异常可一键或按说明恢复）
+    for s in body["scenarios"]:
+        assert s["recover"].strip(), s["id"]
+
+
+def test_fault_apply_mock_api_and_recover(app_client):
+    """N2 预置：apply 改内存态，recover 恢复默认。"""
+    r = app_client.post("/api/fault/apply", json={"scenarioId": "N2-05"})
+    assert r.status_code == 200 and r.json()["applied"] is True
+    assert app.MOCK_STATE.settings.force_status == 500
+    # 恢复默认
+    r = app_client.post("/api/fault/recover", json={"channel": "API_PULL"})
+    assert r.status_code == 200
+    assert app.MOCK_STATE.settings.force_status is None
+    assert app.MOCK_STATE.settings.empty_response is False
+
+
+def test_fault_apply_guide_scenario_is_readonly(app_client):
+    """N1 指引类场景：模拟器不修改 Ai-DOP 配置，返回 guide 说明。"""
+    r = app_client.post("/api/fault/apply", json={"scenarioId": "N1-01"})
+    body = r.json()
+    assert body["applied"] is False and body["guide"] is True
+    assert "人工" in body["note"]
+
+
+def test_fault_apply_override_missing_bizkey(app_client):
+    r = app_client.post("/api/fault/apply", json={"scenarioId": "N2-13"})
+    assert r.json()["applied"] is True
+    path = r.json()["overridePath"]
+    rows = app.MOCK_STATE.overrides[path]
+    assert rows and all("bizKey" not in row for row in rows)
+    app_client.post("/api/fault/recover", json={"channel": "API_PULL"})
+    assert not app.MOCK_STATE.overrides
+
+
+def test_fault_apply_inbound_without_creds_is_tool_blocked_not_reject(app_client, monkeypatch):
+    """N3 apply 缺环境变量时：如实透传 400，不得包装成 expectedReject。
+
+    任务书验收要求「正确拒绝与工具失败区分」：
+    Ai-DOP 预检拦截（409）才算 expectedReject（= 测试通过）；
+    模拟器自身缺凭据（400）是工具阻塞，请求根本没发出。
+    """
+    for name in ("AIDOP_SIM_INBOUND_ACCESS_KEY", "AIDOP_SIM_INBOUND_SECRET"):
+        monkeypatch.delenv(name, raising=False)
+    r = app_client.post("/api/fault/apply", json={"scenarioId": "N3-01", "objectCode": "ITEM"})
+    assert r.status_code == 400
+    body = r.json()
+    assert "expectedReject" not in body
+    assert "缺少环境变量" in body["detail"]
+
+
+class _FakeInboundClient:
+    """拦截发往 Ai-DOP 的请求，捕获构造后的信封/块/坏值。"""
+
+    def __init__(self, *a, **k):
+        self.calls = []
+
+    def op_push(self, entity, envelope, contract_version=None):
+        self.calls.append(("push", entity, json.loads(json.dumps(envelope))))
+        return 202, {"accepted": 2, "rejected": 1}, {"X-Test": "1"}
+
+    def op_bulk(self, entity, bulk_rows, contract_version=None):
+        self.calls.append(("bulk", entity, json.loads(json.dumps(bulk_rows))))
+        return 202, {"accepted": 1, "rejected": 1}, {"X-Test": "1"}
+
+    def case_wrong_field_type(self, entity, envelope, field, bad_value):
+        self.calls.append(("wrong-field-type", entity, field,
+                           json.loads(json.dumps(bad_value))))
+        return 400, {"detail": "type fault"}, {}
+
+
+def _patch_inbound(monkeypatch, fake):
+    monkeypatch.setenv("AIDOP_SIM_INBOUND_ACCESS_KEY", "TEST-AK")
+    monkeypatch.setenv("AIDOP_SIM_INBOUND_SECRET", "TEST-SK")
+    monkeypatch.setattr(app, "InboundClient", lambda *a, **k: fake)
+    monkeypatch.setattr(app.aidop_probe, "precheck_inbound_entity",
+                        lambda client, entity, allowed_hosts, contract_version=None:
+                        {"configState": "READY", "httpStatus": 200})
+
+
+def test_fault_apply_partial_reject_duplicates_row_when_single_sample(app_client, monkeypatch):
+    """N3-12 回归：样例仅 1 行时复制一行再破坏，保证「1 好 1 坏」部分行拒绝可演示。"""
+    fake = _FakeInboundClient()
+    _patch_inbound(monkeypatch, fake)
+    r = app_client.post("/api/fault/apply", json={"scenarioId": "N3-12", "objectCode": "ITEM"})
+    assert r.status_code == 200, r.text
+    op, entity, env = fake.calls[0]
+    assert op == "push" and entity == "MDM_ITEM"
+    rows = env["data"]["list"]
+    assert len(rows) == 2
+    assert "ItemNum" in rows[0]
+    assert "ItemNum" not in rows[1]       # 复制出的坏行缺必填字段
+
+
+def test_fault_apply_bulk_partial_failure_duplicates_chunk_when_single_row(app_client, monkeypatch):
+    """N3-16 回归：样例仅 1 行时复制一个块（新幂等键）再破坏，避免单 chunk 全失败退化。"""
+    fake = _FakeInboundClient()
+    _patch_inbound(monkeypatch, fake)
+    r = app_client.post("/api/fault/apply", json={"scenarioId": "N3-16", "objectCode": "ITEM"})
+    assert r.status_code == 200, r.text
+    op, entity, chunks = fake.calls[0]
+    assert op == "bulk" and entity == "MDM_ITEM"
+    assert len(chunks) == 2
+    keys = [c["idempotencyKey"] for c in chunks]
+    assert len(set(keys)) == 2            # 复制的块带新幂等键
+    assert "ItemNum" in chunks[0]["data"]["list"][0]
+    assert "ItemNum" not in chunks[1]["data"]["list"][0]
+
+
+def test_fault_apply_wrong_field_type_uses_structural_hostile_value(app_client, monkeypatch):
+    """N3-11 回归：坏值必须是结构敌对值（嵌套对象），同类标量可能被宽容转换不触发校验。"""
+    fake = _FakeInboundClient()
+    _patch_inbound(monkeypatch, fake)
+    r = app_client.post("/api/fault/apply", json={"scenarioId": "N3-11", "objectCode": "ITEM"})
+    assert r.status_code == 200, r.text
+    op, entity, field, bad = fake.calls[0]
+    assert op == "wrong-field-type" and entity == "MDM_ITEM" and field == "ItemNum"
+    assert isinstance(bad, dict) and not isinstance(bad, int)
+    assert "ItemNum" not in bad
+
+
+def test_fault_apply_n1_duplicate_bizkey_drops_table_first(app_client, monkeypatch):
+    """N1-07 回归：表已存在时 CREATE IF NOT EXISTS 不重建，必须先 DROP 才能去掉旧唯一键。
+
+    真实 MySQL 实证过：不先 DROP 会残留 uk_ 唯一索引，重复行插入报 1062。
+    """
+    monkeypatch.setenv("AIDOP_SIM_MYSQL_PASSWORD", "x")
+    calls = []
+    monkeypatch.setattr(app.mock_db, "drop_table", lambda t: calls.append(("drop", t)))
+    monkeypatch.setattr(app.mock_db, "init_schema",
+                        lambda tables: calls.append(("init", tables["sim_item"].get("unique_key"))))
+    monkeypatch.setattr(app.mock_db, "insert_rows_raw",
+                        lambda t, rows: calls.append(("insert", len(rows))))
+    r = app_client.post("/api/fault/apply", json={"scenarioId": "N1-07", "objectCode": "ITEM"})
+    assert r.status_code == 200, r.text
+    assert calls[0] == ("drop", "sim_item")
+    assert calls[1] == ("init", False)          # 无唯一键表
+    assert calls[2] == ("insert", 2)            # 两条相同 bizKey
+
+
+def test_fault_recover_db_drops_before_rebuild(app_client, monkeypatch):
+    """恢复回归：N1-07 把表改成无唯一键后，恢复必须先 DROP 再建
+    （init_schema 的 CREATE IF NOT EXISTS 不会改已有表，不 DROP 结构会残留）。"""
+    monkeypatch.setenv("AIDOP_SIM_MYSQL_PASSWORD", "x")
+    calls = []
+    monkeypatch.setattr(app.mock_db, "drop_table", lambda t: calls.append(("drop", t)))
+    monkeypatch.setattr(app.mock_db, "init_schema", lambda tables: calls.append(("init", list(tables))))
+    monkeypatch.setattr(app.mock_db, "upsert_rows", lambda t, rows: calls.append(("upsert", t)))
+    monkeypatch.setattr(app.mock_db, "drop_view", lambda v: calls.append(("dropview", v)))
+    r = app_client.post("/api/fault/recover", json={"channel": "DB_SYNC"})
+    assert r.status_code == 200, r.text
+    idx = calls.index(("drop", "sim_item"))
+    assert calls[idx + 1][0] == "init"       # drop 之后才建表
+    assert "upsert" in [c[0] for c in calls]  # 样例行装载
+    assert ("dropview", "v_slow_sim_item") in calls
+
+
+def test_fault_report_save_list_export_redacted(app_client, monkeypatch):
+    """N4 报告八列 + 脱敏导出。"""
+    monkeypatch.setenv("AIDOP_SIM_INBOUND_ACCESS_KEY", "AK-REAL-SECRET-KEY")
+    r = app_client.post("/api/fault-report/save", json={
+        "scenarioId": "N3-02", "simBehavior": "AccessKey=AK-REAL-SECRET-KEY",
+        "expected": "拒绝", "actualResponse": "401", "aidopFlowStatus": "无流水",
+        "targetChanged": "否", "pass": "是", "note": "正确拒绝"})
+    assert r.status_code == 200, r.text
+    report = r.json()["report"]
+    assert report["channel"] == "API_INBOUND"
+    assert report["scenarioTitle"] == "错误 AccessKey"
+    # 脱敏：AccessKey 只保留后四位
+    assert "AK-REAL-SECRET-KEY" not in report["simBehavior"]
+    assert report["simBehavior"].endswith("-KEY")
+    # 列表与导出
+    lst = app_client.get("/api/fault-report/list").json()["reports"]
+    assert any(item["id"] == report["id"] for item in lst)
+    md = app_client.get(f"/api/fault-report/export/{report['id']}")
+    assert md.status_code == 200
+    assert "异常测试报告" in md.text
+    assert "| 场景 |" in md.text and "| 备注 |" in md.text
+
+
+def test_fault_report_rejects_unknown_scenario(app_client):
+    r = app_client.post("/api/fault-report/save", json={"scenarioId": "N9-99"})
+    assert r.status_code == 400
+
+
+def test_fault_report_save_same_scenario_replaces_in_place(app_client, monkeypatch):
+    """N4 回归：同场景重复保存 = 原位更新（保留原 id），列表不产生重复行。"""
+    monkeypatch.setenv("AIDOP_SIM_INBOUND_ACCESS_KEY", "AK-TEST")
+    base = {"scenarioId": "N3-13", "simBehavior": "v1", "expected": "拒绝",
+            "actualResponse": "400", "aidopFlowStatus": "无流水",
+            "targetChanged": "否", "pass": "是", "note": "第一版"}
+    r1 = app_client.post("/api/fault-report/save", json=base).json()
+    base["note"] = "第二版"
+    r2 = app_client.post("/api/fault-report/save", json=base).json()
+    rep = r2["report"]
+    assert rep["replaced"] is True
+    assert rep["id"] == r1["report"]["id"]      # 原位更新：id 不变
+    rows = [x for x in app_client.get("/api/fault-report/list").json()["reports"]
+            if x["scenarioId"] == "N3-13"]
+    assert len(rows) == 1
+    assert rows[0]["note"] == "第二版"
+
+
+def test_fault_report_save_distinct_scenarios_stay_independent(app_client):
+    """N4 回归：不同场景各自一条，互不覆盖。"""
+    for sid, note in (("N3-14", "甲"), ("N3-15", "乙")):
+        r = app_client.post("/api/fault-report/save", json={
+            "scenarioId": sid, "simBehavior": "x", "expected": "拒绝",
+            "actualResponse": "403", "aidopFlowStatus": "无流水",
+            "targetChanged": "否", "pass": "是", "note": note})
+        assert r.status_code == 200
+        assert r.json()["report"]["replaced"] is False
+    rows = app_client.get("/api/fault-report/list").json()["reports"]
+    assert [x for x in rows if x["scenarioId"] == "N3-14"][0]["note"] == "甲"
+    assert [x for x in rows if x["scenarioId"] == "N3-15"][0]["note"] == "乙"