AiDOPWarehouse/tools/integration-simulator/app.py

"""Ai-DOP 第三方对接模拟器 · 本地控制服务（任务书 P1-B / P1-G）。

- 只绑定 127.0.0.1，页面与 API 仅本机可用；
- Secret/密码/Token 只从环境变量读取，任何 API 不回显；
- 所有对外目标 host 走白名单硬阻断（默认拒绝）；
- 运行记录（RunStore）写入前统一脱敏；
- 内嵌管理 Mock API（providers.mock_api），默认端口 8018，可启停/切鉴权/注入错误；
- 三层报告：代码层 SUPPORTED/UNSUPPORTED、配置层 READY/NOT_*、数据层 NOT_RUN/ACCEPTED/TRANSFORM_PENDING/RUN_FAILED。

启动：python tools/integration-simulator/app.py   （详见 README.md）
"""
from __future__ import annotations

import asyncio
import json
import os
import socket
import sys
import time
import uuid
from pathlib import Path

SIM_ROOT = Path(__file__).resolve().parent
if str(SIM_ROOT) not in sys.path:
    sys.path.insert(0, str(SIM_ROOT))

import uvicorn
from fastapi import FastAPI, HTTPException
from fastapi.responses import FileResponse, Response
from fastapi.staticfiles import StaticFiles
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 fault_catalog, mock_api, mock_db

DEFAULT_CONTROL_PORT = 8017
MAX_RUNS = 200

# 状态枚举（任务书 §4）
CONFIG_STATES = ("READY", "CONFIG_NOT_REGISTERED", "CONFIG_NOT_ENABLED",
                 "GRANT_MISSING", "SOURCE_UNREACHABLE", "RUN_FAILED")


def _env(name: str) -> str:
    return os.environ.get(name, "") or ""


def _secrets() -> list[str]:
    return [v for v in (_env("AIDOP_SIM_INBOUND_SECRET"), _env("AIDOP_SIM_MYSQL_PASSWORD"),
                        _env("AIDOP_SIM_ADMIN_TOKEN")) if v]


def _access_keys() -> list[str]:
    """已知 AccessKey 值：自由文本中出现时只保留后四位（任务书 §24）。"""
    ak = _env("AIDOP_SIM_INBOUND_ACCESS_KEY")
    return [ak] if ak else []


def load_config() -> dict:
    cfg = {
        "aidopBaseUrl": "http://127.0.0.1:5005",
        "mockApiPort": 8018,
        "mysqlDatabase": mock_db.SIM_DB_DEFAULT,
        "allowedTargetHosts": ["127.0.0.1", "localhost"],
    }
    cfg_file = SIM_ROOT / "config.json"
    if cfg_file.exists():
        import json
        with cfg_file.open("r", encoding="utf-8") as f:
            cfg.update(json.load(f))
    return cfg


CONFIG = load_config()
MOCK_STATE = mock_api.MockApiState()
MOCK_APP = mock_api.create_app(MOCK_STATE)


class RunStore:
    def __init__(self, limit: int = MAX_RUNS):
        self.limit = limit
        self.runs: list[dict] = []

    def add(self, run_id: str, object_code: str, channel: str, request_summary: dict,
            response: object, layers: dict, followup_sql: list[str] | None = None,
            gaps: list[str] | None = None, scenario_id: str | None = None,
            nodes: list[dict] | None = None) -> dict:
        entry = {
            "runId": run_id,
            "time": time.strftime("%Y-%m-%dT%H:%M:%S%z"),
            "objectCode": object_code,
            "channel": channel,
            "scenarioId": scenario_id,
            "nodes": nodes,
            "request": redact_obj(request_summary, _secrets(), _access_keys()),
            "response": redact_obj(response, _secrets(), _access_keys()),
            "layers": layers,
            "followupSql": followup_sql or [],
            "knownGaps": gaps or [],
        }
        self.runs.append(entry)
        if len(self.runs) > self.limit:
            del self.runs[: len(self.runs) - self.limit]
        return entry

    def recent(self) -> list[dict]:
        return list(reversed(self.runs[-50:]))


RUNS = RunStore()

app = FastAPI(title="Ai-DOP Integration Simulator", version="1.0.0")


# ---------------------------------------------------------------------------
# Mock API 内嵌启停
# ---------------------------------------------------------------------------

class _MockServerHandle:
    def __init__(self):
        self.server: uvicorn.Server | None = None
        self.task: asyncio.Task | None = None
        self.last_error: str = ""


MOCK_HANDLE = _MockServerHandle()


def _port_in_use(port: int) -> bool:
    """真实探测本机端口是否已被占用。

    不依赖 uvicorn 内部属性：0.30+ 的 Server 已无 startup_error，
    读它会抛 AttributeError 并把已启动的服务误报成 500。
    """
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as probe:
        probe.settimeout(0.3)
        return probe.connect_ex(("127.0.0.1", int(port))) == 0


async def _start_mock_api() -> dict:
    port = int(CONFIG["mockApiPort"])
    if MOCK_HANDLE.task and not MOCK_HANDLE.task.done():
        return {"running": True, "port": port, "alreadyRunning": True}

    # 端口预检必须在建 task 之前：否则失败路径会留下一个已赋值的句柄，
    # 使 _mock_running() 与实际监听状态不一致。
    if _port_in_use(port):
        MOCK_HANDLE.last_error = f"端口 {port} 已被占用（可能是另一个模拟器实例）"
        return {"running": False, "error": MOCK_HANDLE.last_error}

    config = uvicorn.Config(MOCK_APP, host="127.0.0.1", port=port, log_level="warning")
    server = uvicorn.Server(config)
    MOCK_HANDLE.server = server
    MOCK_HANDLE.task = asyncio.get_running_loop().create_task(server.serve())

    # 轮询库的公开状态位，上限 ~2s：比固定 sleep 既不会过早返回未就绪，也不会白等。
    for _ in range(40):
        await asyncio.sleep(0.05)
        if server.started:
            MOCK_HANDLE.last_error = ""
            return {"running": True, "port": port}
        if server.should_exit or MOCK_HANDLE.task.done():
            break

    MOCK_HANDLE.last_error = f"Mock API 在端口 {port} 上启动失败"
    failed_task = MOCK_HANDLE.task
    MOCK_HANDLE.server = None
    MOCK_HANDLE.task = None
    if failed_task and not failed_task.done():
        failed_task.cancel()
    return {"running": False, "error": MOCK_HANDLE.last_error}


async def _stop_mock_api() -> dict:
    if MOCK_HANDLE.server and MOCK_HANDLE.task and not MOCK_HANDLE.task.done():
        MOCK_HANDLE.server.should_exit = True
        try:
            await asyncio.wait_for(MOCK_HANDLE.task, timeout=5)
        except asyncio.TimeoutError:
            MOCK_HANDLE.task.cancel()
        return {"running": False}
    return {"running": False, "alreadyStopped": True}


def _mock_running() -> bool:
    return bool(MOCK_HANDLE.task and not MOCK_HANDLE.task.done() and
                MOCK_HANDLE.server and MOCK_HANDLE.server.started)


# ---------------------------------------------------------------------------
# 基础模型
# ---------------------------------------------------------------------------

class RowsPayload(BaseModel):
    objectCode: str
    runId: str | None = None
    rows: list[dict] | None = None


class InboundExecPayload(BaseModel):
    objectCode: str | None = None
    entityCode: str | None = None
    operation: str = "push"   # schema|push|receipt|bulk|snapshot_open|snapshot_commit|digest
    runId: str | None = None
    rows: list[dict] | None = None
    contractVersion: str | None = None
    # 用例与操作参数
    case: str | None = None   # happy|missing-idem|tamper-body|expired-ts|nonce-replay|same-key-replay|same-key-diff-body|bad-signature
    syncBatchId: str | None = None
    snapshotId: str | None = None
    force: bool = False
    bizDate: str | None = None


class ScenarioPayload(BaseModel):
    scenarioId: str
    runId: str | None = None
    channelOverride: dict[str, str] | None = None


class ReportPayload(BaseModel):
    """全链路报告草稿（任务书 F5 §10）：只保存测试人员填写的摘要。"""
    runId: str
    channel: str                      # DB_SYNC | API_PULL | API_INBOUND
    environment: str = ""             # 环境与测试租户
    objectCode: str = ""              # 对象
    simPrep: str = ""                 # 模拟端准备结果
    manualSteps: str = ""             # 人工执行步骤
    simEvidence: str = ""             # 模拟端证据
    aidopConn: str = ""               # Ai-DOP 测连结果
    flowIds: str = ""                 # 同步/入站流水编号
    stagingCheck: str = ""            # 贴源核验结果
    targetResult: str = ""            # 目标业务结果
    conclusion: str = ""              # 结论
    gaps: str = ""                    # 未完成项
    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（数据库同步）",
    "API_PULL": "方式乙 API_PULL（接口拉取）",
    "API_INBOUND": "方式丙 API_INBOUND（接口推送）",
}

# 报告章节：与任务书 §10 的字段清单一一对应
REPORT_FIELDS = [
    ("runId", "运行编号 runId"),
    ("channel", "集成方式"),
    ("environment", "环境与测试租户"),
    ("objectCode", "对象"),
    ("simPrep", "模拟端准备结果"),
    ("manualSteps", "人工执行步骤"),
    ("simEvidence", "模拟端证据"),
    ("aidopConn", "Ai-DOP 测连结果"),
    ("flowIds", "同步/入站流水编号"),
    ("stagingCheck", "贴源核验结果"),
    ("targetResult", "目标业务结果"),
    ("conclusion", "结论"),
    ("gaps", "未完成项"),
    ("cleanup", "清理提示"),
]


class ReportStore:
    """全链路报告存储（任务书 F5）。

    - 只保存测试人员填写的摘要字段，不保存数据库密码、Secret、JWT、连接串或完整敏感 payload；
    - 保存前统一过 redact_obj，导出内容与内存一致；
    - 内存态、不落盘：控制台重启即清空，避免凭据残留。
    """

    def __init__(self, limit: int = 50):
        self.limit = limit
        self.items: list[dict] = []

    def save(self, fields: dict, secrets: list[str] | None = None,
             access_keys: list[str] | None = None) -> dict:
        entry = {
            "id": uuid.uuid4().hex[:12],
            "createdAt": time.strftime("%Y-%m-%dT%H:%M:%S"),
        }
        entry.update(redact_obj(dict(fields), secrets, access_keys))
        self.items.append(entry)
        if len(self.items) > self.limit:
            del self.items[: len(self.items) - self.limit]
        return entry

    def list(self) -> list[dict]:
        return list(reversed(self.items[-20:]))

    def get(self, report_id: str) -> dict | None:
        return next((r for r in self.items if r["id"] == report_id), None)

    def delete(self, report_id: str) -> bool:
        before = len(self.items)
        self.items = [r for r in self.items if r["id"] != report_id]
        return len(self.items) < before


def build_report_markdown(entry: dict) -> str:
    """报告导出 Markdown（任务书 F5 §10 字段 + 脱敏声明）。"""
    lines = ["# Ai-DOP 第三方系统模拟器 · 全链路测试报告", ""]
    for key, label in REPORT_FIELDS:
        value = str(entry.get(key) or "").strip()
        if key == "channel":
            value = CHANNEL_LABELS.get(value, value)
        lines.append(f"## {label}")
        lines.append(value or "（未填写）")
        lines.append("")
    lines.append("---")
    lines.append(f"报告 ID：{entry.get('id', '')} · 创建时间：{entry.get('createdAt', '')}")
    lines.append("")
    lines.append("> 脱敏说明：本报告已自动脱敏，AccessKey 仅保留后四位；"
                 "数据库密码、Secret、JWT、完整 Authorization 与连接串已遮蔽。")
    return "\n".join(lines)


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)


# ---------------------------------------------------------------------------
# 公共辅助
# ---------------------------------------------------------------------------

def _require_obj(object_code: str) -> dict:
    obj = sa.find_object(object_code)
    if not obj:
        raise HTTPException(status_code=404, detail=f"unknown objectCode: {object_code}")
    return obj


def _build_inbound_client() -> InboundClient:
    base = CONFIG["aidopBaseUrl"]
    try:
        aidop_probe.assert_host_allowed(base, CONFIG["allowedTargetHosts"])
    except aidop_probe.HostNotAllowedError as ex:
        raise HTTPException(status_code=400, detail=str(ex))
    access_key = _env("AIDOP_SIM_INBOUND_ACCESS_KEY")
    secret = _env("AIDOP_SIM_INBOUND_SECRET")
    missing = [n for n, v in (("AIDOP_SIM_INBOUND_ACCESS_KEY", access_key),
                              ("AIDOP_SIM_INBOUND_SECRET", secret)) if not v]
    if missing:
        raise HTTPException(status_code=400,
                            detail=f"缺少环境变量: {', '.join(missing)}（Secret 只从环境变量读取，不落盘）")
    return InboundClient(base, access_key, secret)


def _resolve_entity(payload: InboundExecPayload) -> tuple[dict | None, str]:
    """解析本次执行的对象与 entityCode。仅给 entityCode 时允许无目录对象直发（rows 必填）。"""
    if payload.objectCode:
        obj = _require_obj(payload.objectCode)
    else:
        obj = _require_obj_by_entity(payload.entityCode or "")
    if payload.entityCode:
        entity = payload.entityCode.upper()
    else:
        if not obj:
            raise HTTPException(status_code=400, detail="need objectCode or entityCode")
        inbound = obj.get("apiInbound") or {}
        if not inbound.get("supported"):
            raise HTTPException(status_code=400,
                                detail=f"{obj['objectCode']} 不支持 API_INBOUND: {inbound.get('reason', '无契约')}")
        entity = inbound["entityCode"]
    return obj, entity


def _require_obj_by_entity(entity_code: str) -> dict | None:
    for obj in sa.load_catalog()["objects"]:
        if (obj.get("apiInbound") or {}).get("entityCode") == entity_code.upper():
            return obj
    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}


def _followup_sql(obj: dict, channel: str, run_id: str) -> list[str]:
    prefix = sa.biz_prefix(run_id)
    db = (obj.get("dbSync") or {}).get("table")
    sqls = [
        "SELECT source_code, status, COUNT(*) cnt, MAX(end_time) latest FROM mdp_sync_log "
        "WHERE source_code LIKE 'SIM_MODE3%' GROUP BY source_code, status;",
        "SELECT COUNT(*) FROM mdp_inbound_request WHERE create_time >= CURDATE();",
    ]
    if channel == "DB_SYNC" and db:
        sqls.insert(0, f"-- 模拟库核对：\nSELECT COUNT(*), MAX(sourceUpdatedAt) FROM `{db}` WHERE bizKey LIKE '{prefix}%';")
        sqls.append(f"SELECT * FROM `{db.replace('sim_', 'mdp_stg_sim_')}` WHERE raw_data LIKE '%{prefix}%' LIMIT 20;")
    if channel == "API_INBOUND":
        entity = (obj.get("apiInbound") or {}).get("entityCode")
        sqls.append(f"SELECT status, accepted_rows, rejected_rows, sync_batch_id FROM mdp_inbound_request "
                    f"WHERE entity_code='{entity}' ORDER BY id DESC LIMIT 10;")
    sqls.append("-- 标准层/KPI 下游核验属第二阶段自动对账范围；第一阶段报告 TRANSFORM_PENDING，不猜测成功。")
    return sqls


# ---------------------------------------------------------------------------
# 状态与目录
# ---------------------------------------------------------------------------

@app.get("/api/status")
def status():
    inbound_key = _env("AIDOP_SIM_INBOUND_ACCESS_KEY")
    return {
        "controlPort": DEFAULT_CONTROL_PORT,
        "aidopBaseUrl": CONFIG["aidopBaseUrl"],
        "allowedTargetHosts": CONFIG["allowedTargetHosts"],
        "aidopReachable": None,  # 由 /api/precheck 实测，避免每次状态查询打真实请求
        "mockApi": {"running": _mock_running(), "port": CONFIG["mockApiPort"],
                    "authMode": MOCK_STATE.settings.auth_mode,
                    "forceStatus": MOCK_STATE.settings.force_status,
                    "lastError": MOCK_HANDLE.last_error},
        "mysql": {
            "database": _env("AIDOP_SIM_MYSQL_DATABASE") or mock_db.SIM_DB_DEFAULT,
            "passwordLoaded": bool(_env("AIDOP_SIM_MYSQL_PASSWORD")),
        },
        "inbound": {
            "accessKeyLoaded": bool(inbound_key),
            "accessKeyMasked": mask_access_key(inbound_key),
            "secretLoaded": bool(_env("AIDOP_SIM_INBOUND_SECRET")),
        },
        "adminTokenLoaded": bool(_env("AIDOP_SIM_ADMIN_TOKEN")),
        "objectCounts": _channel_counts(),
    }


def _channel_counts() -> dict:
    counts = {"DB_SYNC": 0, "API_PULL": 0, "API_INBOUND": 0, "total": 0}
    for obj in sa.load_catalog()["objects"]:
        counts["total"] += 1
        if (obj.get("dbSync") or {}).get("supported"):
            counts["DB_SYNC"] += 1
        if (obj.get("apiPull") or {}).get("supported"):
            counts["API_PULL"] += 1
        if (obj.get("apiInbound") or {}).get("supported"):
            counts["API_INBOUND"] += 1
    return counts


@app.get("/api/catalog")
def catalog():
    return sa.load_catalog()


@app.get("/api/scenarios")
def scenarios():
    return sa.load_scenarios()


@app.get("/api/sample/{object_code}")
def get_sample(object_code: str, runId: str | None = None):
    obj = _require_obj(object_code)
    run_id = runId or sa.new_run_id()
    inbound = obj.get("apiInbound") or {}
    result: dict = {"objectCode": obj["objectCode"], "runId": run_id,
                    "canonical": sa.load_canonical_rows(obj),
                    "dbSync": None, "apiPull": None, "apiInbound": None}
    if (obj.get("dbSync") or {}).get("supported"):
        result["dbSync"] = sa.adapt_db_rows(obj, run_id)
    if (obj.get("apiPull") or {}).get("supported"):
        result["apiPull"] = sa.adapt_api_rows(obj, run_id)
    if inbound.get("supported"):
        adapted = sa.adapt_inbound_rows(obj, run_id)
        result["apiInbound"] = {
            "entityCode": inbound["entityCode"],
            "contractVersion": inbound.get("contractVersion", "v1"),
            "rows": adapted["rows"],
            "missingRequired": adapted["missing_required"],
            "envelope": sa.inbound_envelope(adapted["rows"]),
        }
    return result


# ---------------------------------------------------------------------------
# Mock API 控制
# ---------------------------------------------------------------------------

@app.post("/api/mock-api/start")
async def mock_api_start():
    return await _start_mock_api()


@app.post("/api/mock-api/stop")
async def mock_api_stop():
    return await _stop_mock_api()


@app.get("/api/mock-api/settings")
def mock_api_settings():
    s = MOCK_STATE.settings
    return {"running": _mock_running(), "authMode": s.auth_mode, "delaySeconds": s.delay_seconds,
            "forceStatus": s.force_status, "responsePath": s.response_path,
            "apiKeyHeader": s.api_key_header, "overrides": sorted(MOCK_STATE.overrides.keys())}


@app.post("/api/mock-api/control")
def mock_api_control(payload: dict):
    """统一控制入口：auth_mode/token/basic/apikey/delay/force_status/response_path/override。"""
    s = MOCK_STATE.settings
    if "auth_mode" in payload:
        mode = str(payload["auth_mode"]).upper()
        if mode not in ("NONE", "TOKEN", "BASIC", "APIKEY", "OAUTH2"):
            raise HTTPException(status_code=400, detail=f"unknown auth_mode: {mode}")
        s.auth_mode = mode
    for key, attr in (("token", "token"), ("basic_user", "basic_user"),
                      ("basic_password", "basic_password"), ("api_key_header", "api_key_header"),
                      ("api_key_value", "api_key_value"), ("response_path", "response_path")):
        if payload.get(key):
            setattr(s, attr, str(payload[key]))
    if "delay_seconds" in payload:
        s.delay_seconds = max(0.0, float(payload["delay_seconds"]))
    if "force_status" in payload:
        s.force_status = int(payload["force_status"]) if payload["force_status"] else None
    if payload.get("override_path") and isinstance(payload.get("override_rows"), list):
        MOCK_STATE.overrides[payload["override_path"]] = payload["override_rows"]
    if payload.get("clear_overrides"):
        MOCK_STATE.overrides.clear()
    return {"ok": True, "settings": mock_api_settings()}


@app.get("/api/mock-api/audit")
def mock_api_audit():
    return {"entries": redact_obj(MOCK_STATE.audit[-50:], _secrets())}


# ---------------------------------------------------------------------------
# 模拟 MySQL 控制
# ---------------------------------------------------------------------------

@app.post("/api/db/ensure")
def db_ensure(payload: RowsPayload):
    """初始化模拟 schema 并确保对象源表存在（DDL 由样例推断）。"""
    obj = _require_obj(payload.objectCode)
    db_spec = obj.get("dbSync") or {}
    if not db_spec.get("supported"):
        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")
    run_id = payload.runId or sa.new_run_id()
    rows = payload.rows if payload.rows is not None else sa.adapt_db_rows(obj, run_id)
    try:
        result = mock_db.init_schema({db_spec["table"]: {
            "rows": rows, "businessKey": db_spec.get("businessKey"),
            "incrementColumn": db_spec.get("incrementColumn", "sourceUpdatedAt")}})
        return {"ok": True, "runId": run_id, **result}
    except mock_db.BlockedDatabaseError as ex:
        raise HTTPException(status_code=400, detail=str(ex))
    except Exception as ex:
        raise HTTPException(status_code=502, detail=f"MySQL 不可达或执行失败: {type(ex).__name__}: {ex}")


@app.post("/api/db/load")
def db_load(payload: RowsPayload):
    obj = _require_obj(payload.objectCode)
    db_spec = obj.get("dbSync") or {}
    if not db_spec.get("supported"):
        raise HTTPException(status_code=400, detail=f"{obj['objectCode']} 不支持 DB_SYNC")
    run_id = payload.runId or sa.new_run_id()
    rows = payload.rows if payload.rows is not None else sa.adapt_db_rows(obj, run_id)
    try:
        count = mock_db.upsert_rows(db_spec["table"], rows)
        stats = mock_db.table_stats(db_spec["table"], db_spec.get("incrementColumn", "sourceUpdatedAt"))
    except Exception as ex:
        raise HTTPException(status_code=502, detail=f"装载失败（先执行 /api/db/ensure？）: {type(ex).__name__}: {ex}")
    layers = _layers("SUPPORTED", "READY", "ACCEPTED")
    RUNS.add(run_id, obj["objectCode"], "DB_SYNC",
             {"action": "load", "table": db_spec["table"], "rowCount": count},
             stats, layers, _followup_sql(obj, "DB_SYNC", run_id),
             gaps=["Ai-DOP 侧同步须由 MdpDbPullExecutor 真实执行；本操作只装载第三方模拟源库"])
    return {"ok": True, "runId": run_id, "loaded": count, "stats": stats, "layers": layers}


@app.post("/api/db/reset")
def db_reset(payload: RowsPayload):
    obj = _require_obj(payload.objectCode)
    table = (obj.get("dbSync") or {}).get("table")
    if not table:
        raise HTTPException(status_code=400, detail="object has no dbSync table")
    try:
        mock_db.reset_table(table)
        return {"ok": True, "table": table}
    except Exception as ex:
        raise HTTPException(status_code=502, detail=f"{type(ex).__name__}: {ex}")


@app.post("/api/db/update")
def db_update(payload: dict):
    obj = _require_obj(payload.get("objectCode", ""))
    db_spec = obj.get("dbSync") or {}
    table = db_spec.get("table")
    updates = payload.get("updates") or {}
    where = payload.get("where") or {}
    if not table or not updates or not where:
        raise HTTPException(status_code=400, detail="need objectCode/updates/where")
    try:
        affected = mock_db.update_rows(table, updates, where,
                                       db_spec.get("incrementColumn", "sourceUpdatedAt"),
                                       payload.get("newIncrement"))
        return {"ok": True, "table": table, "affected": affected}
    except Exception as ex:
        raise HTTPException(status_code=502, detail=f"{type(ex).__name__}: {ex}")


@app.get("/api/db/stats/{object_code}")
def db_stats(object_code: str):
    obj = _require_obj(object_code)
    db_spec = obj.get("dbSync") or {}
    if not db_spec.get("table"):
        raise HTTPException(status_code=400, detail="object has no dbSync table")
    try:
        return mock_db.table_stats(db_spec["table"], db_spec.get("incrementColumn", "sourceUpdatedAt"))
    except Exception as ex:
        raise HTTPException(status_code=502, detail=f"{type(ex).__name__}: {ex}")


@app.get("/api/db/aidop-config/{object_code}")
def db_aidop_config(object_code: str):
    obj = _require_obj(object_code)
    base = f"http://127.0.0.1:{CONFIG['mockApiPort']}"
    return mock_db.generate_aidop_config(obj, base)


# ---------------------------------------------------------------------------
# 预检与三层报告（P1-G）
# ---------------------------------------------------------------------------

@app.get("/api/precheck")
def precheck():
    base = CONFIG["aidopBaseUrl"]
    reach = aidop_probe.check_reachable(base, CONFIG["allowedTargetHosts"])
    inbound_ready = bool(_env("AIDOP_SIM_INBOUND_ACCESS_KEY") and _env("AIDOP_SIM_INBOUND_SECRET"))
    env_missing = [n for n in ("AIDOP_SIM_INBOUND_ACCESS_KEY", "AIDOP_SIM_INBOUND_SECRET",
                               "AIDOP_SIM_MYSQL_PASSWORD") if not _env(n)]
    return {
        "aidopBaseUrl": base,
        "aidop": reach,
        "mockApi": {"running": _mock_running(), "port": CONFIG["mockApiPort"]},
        "mysql": {"database": _env("AIDOP_SIM_MYSQL_DATABASE") or mock_db.SIM_DB_DEFAULT,
                  "passwordLoaded": bool(_env("AIDOP_SIM_MYSQL_PASSWORD"))},
        "inboundCredentialsLoaded": inbound_ready,
        "envMissing": env_missing,
        "testSources": mock_db.SIM_SOURCES,
        "note": "DB_SYNC/API_PULL 的来源、实体、租户登记须人工在 Ai-DOP 配置页核对（生成清单见 /api/db/aidop-config/{objectCode}）",
    }


@app.get("/api/precheck/inbound")
def precheck_inbound():
    """26 个代码契约逐一 schema 预检（真实签名，只读）。"""
    client = _build_inbound_client()
    codes = sorted({(o.get("apiInbound") or {}).get("entityCode")
                    for o in sa.load_catalog()["objects"]
                    if (o.get("apiInbound") or {}).get("supported")})
    results = aidop_probe.precheck_all_inbound(client, codes, CONFIG["allowedTargetHosts"])
    return {"accessKey": mask_access_key(client.key), "count": len(results), "results": results}


@app.get("/api/precheck/inbound/{entity_code}")
def precheck_inbound_one(entity_code: str):
    client = _build_inbound_client()
    return aidop_probe.precheck_inbound_entity(client, entity_code.upper(), CONFIG["allowedTargetHosts"])


# ---------------------------------------------------------------------------
# API_INBOUND 执行（P1-E）
# ---------------------------------------------------------------------------

@app.post("/api/inbound/exec")
def inbound_exec(payload: InboundExecPayload):
    obj, entity = _resolve_entity(payload)
    client = _build_inbound_client()
    run_id = payload.runId or sa.new_run_id()
    contract_version = payload.contractVersion

    if obj is None:
        # 仅按 entityCode 直发（无目录样例），rows 必填
        rows = payload.rows
        if rows is None and payload.operation in ("push", "bulk"):
            raise HTTPException(status_code=400, detail="entityCode-only exec requires rows")
        envelope = sa.inbound_envelope(rows or [])
    else:
        inbound = obj.get("apiInbound") or {}
        contract_version = contract_version or inbound.get("contractVersion")
        rows = payload.rows if payload.rows is not None else sa.adapt_inbound_rows(obj, run_id)["rows"]
        envelope = sa.inbound_envelope(rows)

    object_code = obj["objectCode"] if obj else entity
    op = payload.operation
    result: dict = {"runId": run_id, "entityCode": entity, "operation": op, "case": payload.case}

    def record(status: int, body, headers, layers: dict, extra: dict | None = None):
        result.update({"httpStatus": status, "response": body, "requestHeaders": headers, **(extra or {})})
        RUNS.add(run_id, object_code, "API_INBOUND",
                 {"operation": op, "case": payload.case, "entityCode": entity,
                  "rowCount": len(rows or []), "contractVersion": contract_version},
                 {"httpStatus": status, "body": body, "headers": headers},
                 layers, _followup_sql(obj or {"objectCode": entity, "apiInbound": {"entityCode": entity}},
                                       "API_INBOUND", run_id))
        return result

    # 预检（配置层）
    pre = aidop_probe.precheck_inbound_entity(client, entity, CONFIG["allowedTargetHosts"], contract_version)
    config_state = pre["configState"]
    result["precheck"] = {"configState": config_state, "httpStatus": pre["httpStatus"]}

    if op == "schema":
        status, body, headers = client.op_schema(entity, contract_version)
        layers = _layers("SUPPORTED", aidop_probe.classify_inbound_precheck(status, body), "NOT_RUN")
        return record(status, body, headers, layers)

    if config_state != "READY" and op in ("push", "bulk", "snapshot_open"):
        layers = _layers("SUPPORTED", config_state, "NOT_RUN")
        raise HTTPException(status_code=409, detail={
            "message": f"配置层 {config_state}：普通推数只对 READY 实体执行（不伪造成功）",
            "precheck": pre, "layers": layers})

    if op == "push":
        case = payload.case or "happy"
        if case == "happy":
            status, body, headers = client.op_push(entity, envelope, contract_version=contract_version)
        elif case == "missing-idem":
            status, body, headers = client.case_missing_idem(entity, envelope)
        elif case == "tamper-body":
            status, body, headers = client.case_tamper_body(entity, envelope)
        elif case == "expired-ts":
            status, body, headers = client.case_expired_ts(entity, envelope)
        elif case == "bad-signature":
            status, body, headers = client.case_bad_signature(entity, envelope)
        elif case == "nonce-replay":
            (s1, d1), (s2, d2) = client.case_nonce_replay(entity, envelope)
            status, body, headers = s2, {"first": [s1, d1], "replay": [s2, d2]}, {}
        elif case == "same-key-replay":
            (s1, d1), (s2, d2) = client.case_same_key_replay(entity, envelope)
            status, body, headers = s2, {"first": [s1, d1], "replay": [s2, d2]}, {}
        elif case == "same-key-diff-body":
            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")
        layers = _layers("SUPPORTED", config_state, data_state)
        return record(status, body, headers, layers,
                      extra={"gaps": ["HTTP 202 仅代表摄取受理；贴源/标准层核验见 followupSql（TRANSFORM_PENDING）"]})

    if op == "receipt":
        if not payload.syncBatchId:
            raise HTTPException(status_code=400, detail="receipt requires syncBatchId")
        status, body, headers = client.op_receipt(payload.syncBatchId)
        layers = _layers("SUPPORTED", config_state, "ACCEPTED" if status == 200 else "RUN_FAILED")
        return record(status, body, headers, layers)

    if op == "bulk":
        # 后端 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)

    if op == "snapshot_open":
        status, body, headers = client.op_snapshot_open(entity)
        layers = _layers("SUPPORTED", config_state, "ACCEPTED" if status in (200, 201) else "RUN_FAILED")
        return record(status, body, headers, layers)

    if op == "snapshot_commit":
        if not payload.snapshotId:
            raise HTTPException(status_code=400, detail="snapshot_commit requires snapshotId")
        status, body, headers = client.op_snapshot_commit(entity, payload.snapshotId, payload.force)
        layers = _layers("SUPPORTED", config_state, "ACCEPTED" if status in (200, 202) else "RUN_FAILED")
        return record(status, body, headers, layers)

    if op == "digest":
        biz_date = payload.bizDate or time.strftime("%Y-%m-%d")
        status, body, headers = client.op_digest(entity, biz_date)
        layers = _layers("SUPPORTED", config_state, "ACCEPTED" if status == 200 else "RUN_FAILED")
        return record(status, body, headers, layers)

    raise HTTPException(status_code=400, detail=f"unknown operation: {op}")


# ---------------------------------------------------------------------------
# 场景执行（P1-F）
# ---------------------------------------------------------------------------

@app.post("/api/scenarios/run")
def scenario_run(payload: ScenarioPayload):
    scenario = next((s for s in sa.load_scenarios()["scenarios"]
                     if s["scenarioId"] == payload.scenarioId), None)
    if not scenario:
        raise HTTPException(status_code=404, detail=f"unknown scenario: {payload.scenarioId}")
    run_id = payload.runId or sa.new_run_id()
    overrides = payload.channelOverride or {}
    nodes_out = []
    for node in scenario["nodes"]:
        obj = sa.find_object(node["objectCode"])
        channel = overrides.get(node["objectCode"], node["channel"])
        node_result = {"objectCode": node["objectCode"], "channel": channel,
                       "relationKey": node.get("relationKey")}
        if not obj:
            node_result.update({"state": "CODE_UNSUPPORTED", "message": "目录中无此对象"})
            nodes_out.append(node_result)
            continue
        spec = {"DB_SYNC": obj.get("dbSync"), "API_PULL": obj.get("apiPull"),
                "API_INBOUND": obj.get("apiInbound")}.get(channel)
        if not spec or not spec.get("supported"):
            reason = (spec or {}).get("reason") or node.get("inboundNote") or "该通道不支持"
            node_result.update({"state": "CODE_UNSUPPORTED", "message": reason})
            nodes_out.append(node_result)
            continue
        try:
            if channel == "DB_SYNC":
                rows = sa.adapt_db_rows(obj, run_id)
                if not _env("AIDOP_SIM_MYSQL_PASSWORD"):
                    raise RuntimeError("缺少 AIDOP_SIM_MYSQL_PASSWORD")
                mock_db.init_schema({spec["table"]: {"rows": rows, "businessKey": spec.get("businessKey"),
                                                     "incrementColumn": spec.get("incrementColumn")}})
                count = mock_db.upsert_rows(spec["table"], rows)
                node_result.update({"state": "ACCEPTED", "loaded": count, "table": spec["table"]})
            elif channel == "API_PULL":
                if not _mock_running():
                    raise RuntimeError("Mock API 未启动（先 POST /api/mock-api/start）")
                rows = sa.adapt_api_rows(obj, run_id)
                MOCK_STATE.overrides[spec["path"]] = rows
                node_result.update({"state": "ACCEPTED", "overridePath": spec["path"], "rowCount": len(rows),
                                    "note": "已在 Mock API 挂载适配样例；由 Ai-DOP MdpApiPullExecutor 真实拉取"})
            else:
                client = _build_inbound_client()
                entity = spec["entityCode"]
                version = node.get("contractVersion") or spec.get("contractVersion")
                pre = aidop_probe.precheck_inbound_entity(client, entity, CONFIG["allowedTargetHosts"], version)
                if pre["configState"] != "READY":
                    node_result.update({"state": pre["configState"],
                                        "message": f"配置层 {pre['configState']}，未推数（不伪造成功）"})
                else:
                    rows = sa.adapt_inbound_rows(obj, run_id)["rows"]
                    status, body, _ = client.op_push(entity, sa.inbound_envelope(rows),
                                                     contract_version=version)
                    node_result.update({"state": "ACCEPTED" if status == 202 else "RUN_FAILED",
                                        "httpStatus": status, "response": body})
        except HTTPException:
            raise
        except aidop_probe.HostNotAllowedError as ex:
            node_result.update({"state": "SOURCE_UNREACHABLE", "message": str(ex)})
        except Exception as ex:
            node_result.update({"state": "RUN_FAILED", "message": f"{type(ex).__name__}: {ex}"})
        nodes_out.append(node_result)

    unsupported = [n["objectCode"] for n in nodes_out if n["state"] == "CODE_UNSUPPORTED"]
    layers = _layers("SUPPORTED",
                     "READY" if all(n["state"] not in ("SOURCE_UNREACHABLE",) for n in nodes_out) else "SOURCE_UNREACHABLE",
                     "ACCEPTED" if all(n["state"] in ("ACCEPTED", "CODE_UNSUPPORTED") for n in nodes_out) else "RUN_FAILED")
    entry = RUNS.add(run_id, ",".join(n["objectCode"] for n in nodes_out), f"SCENARIO:{payload.scenarioId}",
                     {"scenario": scenario["name"]}, {"nodes": nodes_out}, layers,
                     _followup_sql({"objectCode": "SCENARIO", "dbSync": {}}, "SCENARIO", run_id),
                     gaps=([f"以下节点该通道不支持：{', '.join(unsupported)}（链路不完整，不得宣称全链路跑通）"]
                           if unsupported else []),
                     scenario_id=payload.scenarioId, nodes=nodes_out)
    return entry


# ---------------------------------------------------------------------------
# 运行记录
# ---------------------------------------------------------------------------

@app.get("/api/runs")
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）
# ---------------------------------------------------------------------------

@app.post("/api/report/save")
def report_save(payload: ReportPayload):
    if payload.channel not in CHANNELS:
        raise HTTPException(status_code=400, detail=f"unknown channel: {payload.channel}")
    if not payload.runId.strip():
        raise HTTPException(status_code=400, detail="runId required")
    entry = REPORTS.save(payload.model_dump(), _secrets(), _access_keys())
    return {"ok": True, "report": entry}


@app.get("/api/report/list")
def report_list():
    return {"reports": REPORTS.list()}


@app.get("/api/report/evidence/{channel}")
def report_evidence(channel: str):
    """按方式取最近运行记录（已脱敏），供测试人员一键导入「模拟端证据」。"""
    if channel not in CHANNELS:
        raise HTTPException(status_code=404, detail="unknown channel")
    entries = [r for r in RUNS.runs if r["channel"] == channel][-5:]
    return {"entries": list(reversed(entries))}


@app.get("/api/report/export/{report_id}")
def report_export(report_id: str):
    entry = REPORTS.get(report_id)
    if not entry:
        raise HTTPException(status_code=404, detail="report not found")
    md = build_report_markdown(entry)
    filename = f"aidop-sim-report-{entry['runId']}.md"
    return Response(content=md, media_type="text/markdown; charset=utf-8",
                    headers={"Content-Disposition": f'attachment; filename="{filename}"'})


@app.delete("/api/report/{report_id}")
def report_delete(report_id: str):
    if not REPORTS.delete(report_id):
        raise HTTPException(status_code=404, detail="report not found")
    return {"ok": True}


# ---------------------------------------------------------------------------
# 静态页面
# ---------------------------------------------------------------------------

@app.get("/")
def index():
    return FileResponse(SIM_ROOT / "static" / "index.html")


app.mount("/static", StaticFiles(directory=SIM_ROOT / "static"), name="static")


def main() -> None:
    print(f"[simulator] control panel: http://127.0.0.1:{DEFAULT_CONTROL_PORT}/")
    print(f"[simulator] aidop target : {CONFIG['aidopBaseUrl']} (allowed hosts: {CONFIG['allowedTargetHosts']})")
    uvicorn.run(app, host="127.0.0.1", port=DEFAULT_CONTROL_PORT, log_level="info")


if __name__ == "__main__":
    main()