"""风控 Agent 端到端 API 冒烟（真实 HTTP）。

对**运行中的** uvicorn 发真实 HTTP 请求，覆盖 `/api/risk/*` 四端点 +
基金转换 T+1 线（`/api/simulate/trade` convert · `/api/simulate/trade/convert/*`
· `/api/admin/convert/confirm`），并校验 RBAC 角色矩阵、鉴权负例、处置状态机、
适当性归属、AML 扫描幂等。

与 `scripts/dev/sandbox_risk_test.py` 的区别：那个走进程内 `TestClient`，
本脚本走真实 HTTP against uvicorn —— 能捕获「路由没挂上」「进程是旧代码」
这类**进程外**问题（本次实测就靠前置自检抓到 :8000 上跑的是 merge 前的旧进程）。

用法：
    python -m uvicorn app.main:app --host 127.0.0.1 --port 8000
    python scripts/dev/risk_e2e_smoke.py
    python scripts/dev/risk_e2e_smoke.py --only R6
"""

from __future__ import annotations

import argparse
import json
import sys
from dataclasses import dataclass, field
from datetime import date, datetime, timedelta
from pathlib import Path

import httpx

# 直接 `python scripts/dev/risk_e2e_smoke.py` 时 sys.path[0] 是脚本目录，
# 签发令牌 / 查库需要仓库根目录在 path 上。
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))

# ── 演示账号（与 app/gateway/jwt_service.py:DEFAULT_ROLES_BY_ACTOR 对齐）──
OFFICER = "STAFF-30001"  # risk_officer + risk_demo
MANAGER = "STAFF-31001"  # risk_manager（无 risk_demo）
COMPLIANCE = "STAFF-40001"  # compliance
ADVISOR = "STAFF-10086"  # advisor
CUSTOMER = "CUST-9527"  # customer 本人

#: 转换受理 —— 合法对：转出/转入**同管理人同 TA**（华夏模拟基金 · TA-CN-001）。
#: CUST-1001 实测持有 PROD-110022 共 30000 份。
CONVERT_FROM = "PROD-110022"
CONVERT_TO = "PROD-000001"
CONVERT_CUSTOMER = "CUST-1001"
CONVERT_QTY = 30000

#: 前端 RiskConvertPage.tsx:16-27「演示全转」预置的转入方 —— 属**易方模拟基金 ·
#: TA-CN-002**，与转出方跨主体，服务端按「同销售机构+同管理人+同 TA」拒绝。
#: 保留此常量用于固化该预置缺陷的探针（见 r6）。
CONVERT_TO_PRESET = "PROD-161725"

#: 受理用「有份额余量」的客户 —— CUST-1001 仅持 30000 份 PROD-110022，
#: 首笔全转后可用即归零（服务端「已扣除在途转换占用」），后续用例会假 400。
#: CUST-9527 持 PROD-110022 共 80000 份（同为华夏/TA-CN-001），留足余量。
CONVERT_CUSTOMER_FUNDED = "CUST-9527"
CONVERT_QTY_MAIN = 5000
CONVERT_QTY_IDEM = 2000
CONVERT_QTY_CANCEL = 3000

PASS, FAIL, SKIP = "PASS", "FAIL", "SKIP"

#: 运行中后端必须具备的 convert 路径（否则是旧进程，全部转换用例会假 FAIL）
REQUIRED_CONVERT_PATHS = {
    "/api/admin/convert/confirm",
    "/api/simulate/trade/convert/{convert_group_id}",
    "/api/simulate/trade/convert/{convert_group_id}/cancel",
}


@dataclass
class Result:
    group: str
    name: str
    status: str
    expected: str
    actual: str
    evidence: str = ""


@dataclass
class Ctx:
    client: httpx.Client
    h: dict[str, dict] = field(default_factory=dict)  # 角色 → 请求头
    state: dict = field(default_factory=dict)


RESULTS: list[Result] = []


def record(group: str, name: str, status: str, expected: str, actual: str, evidence: str = "") -> None:
    RESULTS.append(Result(group, name, status, expected, actual, evidence[:600]))
    mark = {PASS: "  ok ", FAIL: " FAIL", SKIP: " skip"}[status]
    print(f"[{mark}] {group}  {name}  | {actual}")


def ok(group: str, name: str, expected: str, actual: str, evidence: str = "") -> None:
    record(group, name, PASS, expected, actual, evidence)


def bad(group: str, name: str, expected: str, actual: str, evidence: str = "") -> None:
    record(group, name, FAIL, expected, actual, evidence)


def skip(group: str, name: str, reason: str) -> None:
    record(group, name, SKIP, "-", reason)


def expect_status(group: str, name: str, resp: httpx.Response, want: int) -> bool:
    """断言 HTTP 状态码；返回是否通过，便于调用方决定后续依赖用例是否 SKIP。"""
    if resp.status_code == want:
        ok(group, name, f"HTTP {want}", f"HTTP {resp.status_code}")
        return True
    bad(group, name, f"HTTP {want}", f"HTTP {resp.status_code}", resp.text[:400])
    return False


def expect_code(group: str, name: str, resp: httpx.Response, want_status: int, want_code: str) -> bool:
    """断言 HTTP 状态码 **且** 业务 error_code。"""
    got_code = err_code(resp)
    if resp.status_code == want_status and got_code == want_code:
        ok(group, name, f"HTTP {want_status} {want_code}", f"HTTP {resp.status_code} {got_code}")
        return True
    bad(
        group,
        name,
        f"HTTP {want_status} {want_code}",
        f"HTTP {resp.status_code} {got_code}",
        resp.text[:400],
    )
    return False


def body(resp: httpx.Response) -> dict:
    try:
        return resp.json()
    except Exception:
        return {}


def err_code(resp: httpx.Response) -> str | None:
    b = body(resp)
    # 平台 ApiError → 顶层 error_code；AdvisorAppError → data.error_code
    return b.get("error_code") or (b.get("data") or {}).get("error_code")


# --------------------------------------------------------------------------
# 前置自检
# --------------------------------------------------------------------------
def preflight(ctx: Ctx, base_url: str) -> bool:
    print(f"== 风控 Agent 端到端冒烟 @ {base_url} ==\n")
    failures: list[str] = []

    try:
        r = ctx.client.get("/health", timeout=10)
        if r.status_code != 200 or body(r).get("status") != "ok":
            failures.append(f"/health 未就绪: HTTP {r.status_code} {r.text[:120]}")
        else:
            print(f"  [ok]   /health → {body(r).get('status')} (env={body(r).get('env')})")
    except Exception as exc:  # noqa: BLE001
        failures.append(f"/health 不可达: {exc}")

    # 关键：确认后端进程跑的是**当前**代码。旧进程缺 convert 路由，
    # 会让 R6 整组假 FAIL —— 本次实测就靠这条抓到 :8000 上的 merge 前旧进程。
    try:
        spec = ctx.client.get("/openapi.json", timeout=20).json()
        paths = set(spec.get("paths", {}))
        missing = REQUIRED_CONVERT_PATHS - paths
        if missing:
            failures.append(
                f"后端缺 convert 路由（旧进程？需重启）: {sorted(missing)}；"
                f"当前共 {len(paths)} 条路径"
            )
        else:
            print(f"  [ok]   openapi 含 3 条 convert 路径（共 {len(paths)} 条）")
    except Exception as exc:  # noqa: BLE001
        failures.append(f"/openapi.json 不可读: {exc}")

    if failures:
        print("\n前置自检失败：")
        for f in failures:
            print(f"  - {f}")
        return False

    # 令牌：演示账号走真实登录接口（与前端同路径），特殊角色走 dev 签发
    try:
        ctx.h["officer"] = login(ctx, OFFICER)
        ctx.h["manager"] = login(ctx, MANAGER)
        ctx.h["compliance"] = login(ctx, COMPLIANCE)
        ctx.h["advisor"] = login(ctx, ADVISOR)
        print("  [ok]   演示账号令牌 4/4")
    except Exception as exc:  # noqa: BLE001
        print(f"\n前置自检失败：登录失败 {exc}")
        return False
    return True


def login(ctx: Ctx, actor_id: str) -> dict:
    """走真实登录接口，返回**不含** X-Agent-Type 的基础头。"""
    r = ctx.client.post("/api/auth/login", json={"actor_id": actor_id, "token_type": "staff"})
    r.raise_for_status()
    b = body(r)
    token = b.get("access_token") or (b.get("data") or {}).get("access_token")
    if not token:
        raise RuntimeError(f"登录响应无 access_token: {r.text[:200]}")
    return {"Authorization": f"Bearer {token}"}


def risk_h(base: dict) -> dict:
    """风控线请求头：Bearer + 必需的 X-Agent-Type。"""
    return {**base, "X-Agent-Type": "risk"}


def db_query(sql: str, db: str = "core") -> list[tuple]:
    """只读查库，用于交叉印证（失败返回空表，不阻塞用例）。"""
    try:
        import pymysql

        from app.config.settings import settings

        name = settings.mysql_core_database if db == "core" else settings.mysql_database
        conn = pymysql.connect(
            host=settings.mysql_host,
            port=int(settings.mysql_port),
            user=settings.mysql_user,
            password=settings.mysql_password,
            database=name,
            charset="utf8mb4",
        )
        try:
            with conn.cursor() as cur:
                cur.execute(sql)
                return cur.fetchall()
        finally:
            conn.close()
    except Exception as exc:  # noqa: BLE001
        print(f"  (查库跳过: {exc})")
        return []


# --------------------------------------------------------------------------
# R1 鉴权负例
# --------------------------------------------------------------------------
def r1(ctx: Ctx) -> None:
    G = "R1"
    path = "/api/risk/alerts"

    # dev 环境 jwt_public_key_path 为空 → 无 token 落到 debug 头兜底通道，
    # 故错误码是 MISSING_DEBUG_HEADERS 而非 MISSING_BEARER（生产口径见 deps.py:259-271）。
    r = ctx.client.get(path)
    expect_code(G, "无 token → 401", r, 401, "AUTH_401_MISSING_DEBUG_HEADERS")

    r = ctx.client.get(path, headers=ctx.h["officer"])  # 有 token，无 X-Agent-Type
    expect_code(G, "有 token 无 X-Agent-Type → 401", r, 401, "AUTH_401_MISSING_AGENT_TYPE")

    r = ctx.client.get(path, headers={**ctx.h["officer"], "X-Agent-Type": "foo"})
    expect_code(G, "X-Agent-Type 越域 → 400", r, 400, "BAD_REQUEST")

    r = ctx.client.get(path, headers={**ctx.h["advisor"], "X-Agent-Type": "risk"})
    expect_code(G, "advisor 冒充 risk → 403", r, 403, "AUTH_403_AGENT_MISMATCH")

    # AGENT_ACCESS_MATRIX["advisor"].roles 含 compliance（deps.py:55），
    # 故 compliance 声明 advisor 域是**合法**的，不作为负例。
    # 真正的负例要用不在该域内的角色：risk_officer ∉ advisor 域。
    r = ctx.client.get(path, headers={**ctx.h["officer"], "X-Agent-Type": "advisor"})
    expect_code(G, "risk_officer 声明 advisor 域 → 403", r, 403, "AUTH_403_AGENT_MISMATCH")

    r = ctx.client.get(path, headers={**ctx.h["compliance"], "X-Agent-Type": "analyst"})
    expect_status(G, "compliance 声明 analyst 域 → 200（矩阵允许）", r, 200)


# --------------------------------------------------------------------------
# R5 AML 扫描（先跑：为 R2 统计 / R3 处置造出 pending 预警）
# --------------------------------------------------------------------------
def r5(ctx: Ctx) -> None:
    G = "R5"
    path = "/api/risk/aml/scan"

    r = ctx.client.post(path, headers=risk_h(ctx.h["officer"]))
    if expect_status(G, "officer 执行 AML 扫描 → 200", r, 200):
        d = body(r)
        for key in ("scanned", "hit_customers", "alerts", "skipped_existing"):
            if key in d:
                ok(G, f"扫描返回含 {key}", f"{key} 存在", f"{key}={d[key]!r}")
            else:
                bad(G, f"扫描返回含 {key}", f"{key} 存在", "缺失", json.dumps(d, ensure_ascii=False)[:300])
        ctx.state["aml_scan"] = d

    r2 = ctx.client.post(path, headers=risk_h(ctx.h["officer"]))
    if r2.status_code == 200:
        ok(G, "重复扫描 → 200（幂等）", "HTTP 200", "HTTP 200")
    else:
        bad(G, "重复扫描 → 200（幂等）", "HTTP 200", f"HTTP {r2.status_code}", r2.text[:300])

    r = ctx.client.post(path, headers=risk_h(ctx.h["manager"]))
    expect_code(G, "manager → 403", r, 403, "AUTH_403_ROLE")

    r = ctx.client.post(path, headers=risk_h(ctx.h["compliance"]))
    expect_code(G, "compliance → 403", r, 403, "AUTH_403_ROLE")

    r = ctx.client.post(path, headers=risk_h(ctx.h["advisor"]))
    expect_code(G, "advisor → 403", r, 403, "AUTH_403_AGENT_MISMATCH")


# --------------------------------------------------------------------------
# R2 台账 RBAC 矩阵
# --------------------------------------------------------------------------
def r2(ctx: Ctx) -> None:
    G = "R2"
    path = "/api/risk/alerts"

    for role, want in (("officer", 200), ("manager", 200), ("compliance", 200)):
        r = ctx.client.get(path, headers=risk_h(ctx.h[role]))
        if expect_status(G, f"{role} → 200", r, want):
            d = body(r)
            ctx.state[f"alerts_{role}"] = d
            if role == "officer":
                items = d.get("items")
                if isinstance(items, list):
                    ok(G, "返回 items 为列表", "list", f"len={len(items)}")
                else:
                    bad(G, "返回 items 为列表", "list", type(items).__name__)
                stats = d.get("stats") or {}
                for k in ("pending_review_count", "today_pending_count"):
                    if isinstance(stats.get(k), int) and stats[k] >= 0:
                        ok(G, f"stats.{k} 为非负整数", "int>=0", f"{stats[k]}")
                    else:
                        bad(G, f"stats.{k} 为非负整数", "int>=0", repr(stats.get(k)), json.dumps(stats)[:200])
                if d.get("disclaimer"):
                    ok(G, "含 disclaimer", "非空", str(d["disclaimer"])[:60])
                else:
                    bad(G, "含 disclaimer", "非空", "缺失")

    # advisor ∉ risk 域（deps.py:56）→ 在**矩阵层**即被拦，早于路由内角色判定，
    # 故错误码是 AGENT_MISMATCH 而非 ROLE。
    r = ctx.client.get(path, headers=risk_h(ctx.h["advisor"]))
    expect_code(G, "advisor → 403（矩阵层拦截）", r, 403, "AUTH_403_AGENT_MISMATCH")

    # compliance 强制 aml 视图
    r = ctx.client.get(path, headers=risk_h(ctx.h["compliance"]))
    if r.status_code == 200:
        items = body(r).get("items") or []
        types = {i.get("alert_type") for i in items}
        if types <= {"aml"}:
            ok(G, "compliance 台账仅 aml", "alert_type ⊆ {aml}", f"实际 {sorted(t for t in types if t)}")
        else:
            bad(G, "compliance 台账仅 aml", "alert_type ⊆ {aml}", f"实际 {sorted(t for t in types if t)}")

    # 分页与参数校验
    r = ctx.client.get(path, headers=risk_h(ctx.h["officer"]), params={"page_size": 1})
    if r.status_code == 200:
        n = len(body(r).get("items") or [])
        (ok if n <= 1 else bad)(G, "page_size=1 生效", "items<=1", f"items={n}")
    r = ctx.client.get(path, headers=risk_h(ctx.h["officer"]), params={"page_size": 101})
    expect_code(G, "page_size=101 → 422", r, 422, "REQUEST_VALIDATION_FAILED")
    # 观察项（非缺陷）：status 无枚举校验，未知值按 SQL 等值匹配 → 静默返回空集，
    # 而非 422。前端 RiskAlertsPage 用固定 Select，用户无法输入非法值，故影响低。
    r = ctx.client.get(path, headers=risk_h(ctx.h["officer"]), params={"status": "nope"})
    if r.status_code == 200 and len(body(r).get("items") or []) == 0:
        ok(G, "非法 status → 200 空集（未做枚举校验）", "HTTP 200 items=0", "HTTP 200 items=0")
    else:
        bad(G, "非法 status → 200 空集（未做枚举校验）", "HTTP 200 items=0",
            f"HTTP {r.status_code} items={len(body(r).get('items') or [])}")


# --------------------------------------------------------------------------
# R3 处置状态机
# --------------------------------------------------------------------------
def ensure_pending_alert(ctx: Ctx) -> str | None:
    """取一个 pending_review 的 alert_id；没有就先造（AML 扫描 → 模拟交易）。"""
    def fetch() -> str | None:
        rows = db_query("SELECT alert_id FROM risk_alert WHERE status='pending_review' ORDER BY created_at LIMIT 1", "agent")
        return rows[0][0] if rows else None

    aid = fetch()
    if aid:
        return aid
    ctx.client.post("/api/risk/aml/scan", headers=risk_h(ctx.h["officer"]))
    aid = fetch()
    if aid:
        return aid
    # 兜底：模拟一笔大额交易触发预警
    ctx.client.post(
        "/api/simulate/trade",
        headers=risk_h(ctx.h["officer"]),
        json={"customer_id": CUSTOMER, "product_id": "PROD-005827",
              "trade_type": "subscribe", "amount": 900000},
    )
    return fetch()


def r3(ctx: Ctx) -> None:
    G = "R3"
    aid = ensure_pending_alert(ctx)
    if not aid:
        skip(G, "处置状态机（整组）", "取不到 pending_review 预警，无法构造处置用例")
        return
    ctx.state["alert_id"] = aid
    path = f"/api/risk/alerts/{aid}/handle"

    r = ctx.client.post(path, headers=risk_h(ctx.h["manager"]), json={"handler_result": "confirmed_normal"})
    expect_code(G, "manager 处置 → 403", r, 403, "AUTH_403_ROLE")

    r = ctx.client.post(path, headers=risk_h(ctx.h["officer"]), json={"handler_result": "nope"})
    expect_code(G, "handler_result 非法 → 422", r, 422, "REQUEST_VALIDATION_FAILED")

    r = ctx.client.post(path, headers=risk_h(ctx.h["officer"]), json={"handler_result": "confirmed_normal", "handler_comment": "x" * 513})
    expect_code(G, "handler_comment 超 512 → 422", r, 422, "REQUEST_VALIDATION_FAILED")

    r = ctx.client.post("/api/risk/alerts/ALERT-NOPE-999/handle", headers=risk_h(ctx.h["officer"]), json={"handler_result": "confirmed_normal"})
    expect_code(G, "不存在 alert → 404", r, 404, "NOT_FOUND")

    r = ctx.client.post(path, headers=risk_h(ctx.h["officer"]), json={"handler_result": "confirmed_normal", "handler_comment": "E2E 冒烟"})
    if expect_status(G, "officer 处置 → 200", r, 200):
        d = body(r)
        status = d.get("status")
        if status and status != "pending_review":
            ok(G, "状态由 pending_review 迁移", "status != pending_review", f"status={status}")
        else:
            bad(G, "状态由 pending_review 迁移", "status != pending_review", f"status={status}", json.dumps(d, ensure_ascii=False)[:300])

    r = ctx.client.post(path, headers=risk_h(ctx.h["officer"]), json={"handler_result": "confirmed_normal"})
    expect_code(G, "重复处置 → 409", r, 409, "STATE_CONFLICT")


# --------------------------------------------------------------------------
# R4 适当性
# --------------------------------------------------------------------------
def r4(ctx: Ctx) -> None:
    G = "R4"
    path = "/api/risk/suitability/check"

    before = db_query("SELECT COUNT(*) FROM audit_log WHERE event_type='suitability_check'", "agent")

    r = ctx.client.post(path, headers=risk_h(ctx.h["officer"]), json={"customer_id": CUSTOMER, "product_id": "PROD-161725"})
    if expect_status(G, "officer 校验 → 200", r, 200):
        d = body(r)
        blocked = d.get("blocked")
        if isinstance(blocked, bool):
            ok(G, "返回 blocked 布尔", "bool", f"blocked={blocked}")
        else:
            bad(G, "返回 blocked 布尔", "bool", repr(blocked), json.dumps(d, ensure_ascii=False)[:300])
        if blocked:
            for k in ("advice", "notice"):
                (ok if d.get(k) else bad)(G, f"阻断时含 {k}", "非空", str(d.get(k))[:80])
        else:
            skip(G, "阻断文案", "该客户/产品未阻断，跳过 advice/notice 断言")

    r = ctx.client.post(path, headers=risk_h(ctx.h["manager"]), json={"customer_id": CUSTOMER, "product_id": "PROD-161725"})
    expect_code(G, "manager → 403 SCOPE", r, 403, "AUTH_403_SCOPE")

    r = ctx.client.post(path, headers=risk_h(ctx.h["compliance"]), json={"customer_id": CUSTOMER, "product_id": "PROD-161725"})
    expect_code(G, "compliance → 403 SCOPE", r, 403, "AUTH_403_SCOPE")

    # 顾问**经 JWT 通道**根本无法到达本端点：advisor ∉ risk 域（deps.py:56），
    # 在矩阵层即被拦。`sandbox_risk_test.py` 断言的 NOT_ASSIGNED/NOT_OWNER 走的是
    # dev 的 X-Debug-Role 兜底通道（不过矩阵），两通道口径不同 —— 见报告「通道差异」。
    r = ctx.client.post(path, headers=risk_h(ctx.h["advisor"]), json={"customer_id": "CUST-1001", "product_id": "PROD-161725"})
    expect_code(G, "advisor（JWT 通道）→ 403 矩阵拦截", r, 403, "AUTH_403_AGENT_MISMATCH")

    after = db_query("SELECT COUNT(*) FROM audit_log WHERE event_type='suitability_check'", "agent")
    if before and after:
        delta = after[0][0] - before[0][0]
        if delta >= 1:
            ok(G, "每次调用落 audit_log", "Δ>=1", f"Δ={delta}")
        else:
            bad(G, "每次调用落 audit_log", "Δ>=1", f"Δ={delta}")


# --------------------------------------------------------------------------
# R6 基金转换线（本轮重点）
# --------------------------------------------------------------------------
def r6(ctx: Ctx) -> None:
    G = "R6"
    accept_path = "/api/simulate/trade"

    # ── 探针：前端「演示全转」预置对（110022→161725）跨主体，必 400 ──
    # RiskConvertPage.tsx:16-27 的 ACCEPT_PRESETS 把转入方写成 PROD-161725
    # （易方模拟基金/TA-CN-002），与转出方 PROD-110022（华夏/TA-CN-001）不同管理人，
    # 服务端按「同销售机构+同管理人+同 TA」拒绝 → 页面上点「提交受理」必得 400。
    r = ctx.client.post(
        accept_path,
        headers=risk_h(ctx.h["officer"]),
        json={
            "customer_id": CONVERT_CUSTOMER,
            "trade_type": "convert",
            "from_product_id": CONVERT_FROM,
            "to_product_id": CONVERT_TO_PRESET,
            "qty": CONVERT_QTY,
        },
    )
    expect_code(G, "前端预置对（跨管理人）→ 400", r, 400, "CROSS_ENTITY_NOT_SUPPORTED")

    # ── 受理：合法（同管理人同 TA）→ 202 ──
    payload = {
        "customer_id": CONVERT_CUSTOMER_FUNDED,
        "trade_type": "convert",
        "from_product_id": CONVERT_FROM,
        "to_product_id": CONVERT_TO,
        "qty": CONVERT_QTY_MAIN,
    }
    r = ctx.client.post(accept_path, headers=risk_h(ctx.h["officer"]), json=payload)
    gid = None
    if r.status_code == 202:
        d = body(r)
        gid = d.get("convert_group_id")
        ctx.state["convert_gid"] = gid
        ok(G, "convert 合法受理 → 202", "HTTP 202", "HTTP 202")
        if isinstance(gid, str) and gid.startswith("CNV-"):
            ok(G, "回执含 convert_group_id", "^CNV-...", gid)
        else:
            bad(G, "回执含 convert_group_id", "^CNV-...", repr(gid), json.dumps(d, ensure_ascii=False)[:300])
        st = d.get("status")
        if st in ("accepted", "processing"):
            ok(G, "受理状态 accepted/processing", "accepted|processing", f"status={st}")
        else:
            bad(G, "受理状态 accepted/processing", "accepted|processing", repr(st), json.dumps(d, ensure_ascii=False)[:300])
    elif r.status_code == 200 and body(r).get("blocked"):
        ok(G, "convert 合法受理 → 202", "HTTP 202", "HTTP 200 + blocked=true（适当性阻断，非受理失败）",
           json.dumps(body(r), ensure_ascii=False)[:300])
        skip(G, "convert 回执 / 查询 / 撤单", "受理被适当性阻断，后续依赖用例跳过")
        return
    else:
        bad(G, "convert 合法受理 → 202", "HTTP 202", f"HTTP {r.status_code}", r.text[:400])
        skip(G, "convert 回执 / 查询 / 撤单", "受理未成功")
        return

    # ── 幂等：同 client_request_id 重放 → 同 group_id ──
    # 注意：幂等键必须**每次运行唯一**，否则重跑会命中上一轮的单、拿不到「首次受理」语义。
    stamp = datetime.now().strftime("%Y%m%d%H%M%S")
    idem_payload = {**payload, "qty": CONVERT_QTY_IDEM, "client_request_id": f"E2E-IDEM-{stamp}"}
    r1 = ctx.client.post(accept_path, headers=risk_h(ctx.h["officer"]), json=idem_payload)
    r2 = ctx.client.post(accept_path, headers=risk_h(ctx.h["officer"]), json=idem_payload)
    if r1.status_code == 202 and r2.status_code == 202:
        b1, b2 = body(r1), body(r2)
        g1, g2 = b1.get("convert_group_id"), b2.get("convert_group_id")
        if g1 and g1 == g2:
            ok(G, "同幂等键重放 → 同 group_id", "gid 相同", f"{g1}")
        else:
            bad(G, "同幂等键重放 → 同 group_id", "gid 相同", f"{g1} vs {g2}")
        if b1.get("idempotent") is False and b2.get("idempotent") is True:
            ok(G, "首次 idempotent=false / 重放 =true", "false → true",
               f"{b1.get('idempotent')} → {b2.get('idempotent')}")
        else:
            bad(G, "首次 idempotent=false / 重放 =true", "false → true",
                f"{b1.get('idempotent')} → {b2.get('idempotent')}",
                json.dumps(b2, ensure_ascii=False)[:300])
    else:
        bad(G, "同幂等键重放 → 202", "两次均 202", f"{r1.status_code} / {r2.status_code}",
            (r1.text[:200] + " | " + r2.text[:200]))

    # ── 权限矩阵 ──
    r = ctx.client.post(accept_path, headers=risk_h(ctx.h["manager"]), json={**payload, "client_request_id": None})
    expect_code(G, "manager（无 risk_demo）→ 403", r, 403, "AUTH_403_ROLE")

    r = ctx.client.post(accept_path, headers=risk_h(ctx.h["advisor"]), json=payload)
    expect_code(G, "advisor 冒充 risk → 403", r, 403, "AUTH_403_AGENT_MISMATCH")

    # ── 查询 ──
    if gid:
        r = ctx.client.get(f"/api/simulate/trade/convert/{gid}", headers=risk_h(ctx.h["officer"]))
        if expect_status(G, "查询受理单 → 200", r, 200):
            d = body(r)
            for k in ("status", "accept_date"):
                (ok if d.get(k) is not None else bad)(G, f"查询返回 {k}", "非空", str(d.get(k)))
            for k in ("out_nav", "convert_amount", "in_qty"):
                if k not in d:
                    ok(G, f"未确认单不含 {k}", "字段不存在", "不存在（Q2 契约）")
                else:
                    bad(G, f"未确认单不含 {k}", "字段不存在", f"实际存在 {k}={d.get(k)!r}")

        r = ctx.client.get("/api/simulate/trade/convert/CNV-19700101-DEADBEEF", headers=risk_h(ctx.h["officer"]))
        expect_status(G, "查询不存在单 → 404", r, 404)

        # ── 撤单：必须用**独立新单**，且必须在 confirm 之前 ──
        # confirm_batch 会把 accepted 单推进到 nav_pending，而 nav_pending 不可撤（409）。
        # 故这里新开一张单专供撤单，主单 gid 保持 accepted 留给后面的 confirm 扫描。
        cancel_payload = {
            **payload,
            "qty": CONVERT_QTY_CANCEL,
            "client_request_id": f"E2E-CANCEL-{stamp}",
        }
        rc = ctx.client.post(accept_path, headers=risk_h(ctx.h["officer"]), json=cancel_payload)
        cgid = body(rc).get("convert_group_id") if rc.status_code == 202 else None
        if not cgid:
            skip(G, "撤单 / 重复撤单", f"撤单用新单未受理（HTTP {rc.status_code}）")
        else:
            r = ctx.client.post(f"/api/simulate/trade/convert/{cgid}/cancel", headers=risk_h(ctx.h["officer"]))
            if r.status_code == 200:
                d = body(r)
                st = d.get("status")
                if st == "cancelled":
                    ok(G, "撤单 → status=cancelled", "cancelled", f"status={st}")
                else:
                    bad(G, "撤单 → status=cancelled", "cancelled", f"status={st}", json.dumps(d, ensure_ascii=False)[:300])
                r2 = ctx.client.post(f"/api/simulate/trade/convert/{cgid}/cancel", headers=risk_h(ctx.h["officer"]))
                if r2.status_code == 409:
                    ok(G, "重复撤单 → 409", "HTTP 409", f"HTTP 409 {err_code(r2)}")
                else:
                    bad(G, "重复撤单 → 409", "HTTP 409", f"HTTP {r2.status_code}", r2.text[:300])
            else:
                bad(G, "撤单 → 200", "HTTP 200", f"HTTP {r.status_code}", r.text[:400])

        # ── 确认批处理：区间内 vs 区间外 ──
        nav_rows = db_query("SELECT MAX(nav_date) FROM core_product_nav")
        nav_max = nav_rows[0][0] if nav_rows else None
        if nav_max:
            inside = nav_max.isoformat()
            outside = (nav_max + timedelta(days=4)).isoformat()
            for label, dt in (("区间内", inside), ("区间外", outside)):
                r = ctx.client.post("/api/admin/convert/confirm", headers=risk_h(ctx.h["officer"]), params={"accept_date": dt})
                if r.status_code == 200:
                    d = body(r)
                    ok(G, f"确认批处理 {label} {dt} → 200", "HTTP 200",
                       f"scanned={d.get('scanned')} confirmed={d.get('confirmed')} nav_pending={d.get('nav_pending')}",
                       json.dumps(d, ensure_ascii=False)[:400])
                else:
                    bad(G, f"确认批处理 {label} {dt} → 200", "HTTP 200", f"HTTP {r.status_code}", r.text[:300])

        # ── 确认参数校验与权限 ──
        # accept_date 非法格式由 FastAPI 查询参数校验拦下（早于 handler）→ 422，
        # 非 handler 内的 400。实测口径，勿按 400 断言。
        r = ctx.client.post("/api/admin/convert/confirm", headers=risk_h(ctx.h["officer"]), params={"accept_date": "2026/09/15"})
        expect_code(G, "accept_date 格式非法 → 422", r, 422, "REQUEST_VALIDATION_FAILED")

        r = ctx.client.post("/api/admin/convert/confirm", headers=risk_h(ctx.h["officer"]))
        expect_status(G, "缺 accept_date → 422", r, 422)

        r = ctx.client.post("/api/admin/convert/confirm", headers=risk_h(ctx.h["manager"]), params={"accept_date": inside})
        expect_code(G, "manager 确认 → 403", r, 403, "AUTH_403_ROLE")

        r = ctx.client.post("/api/admin/convert/confirm", headers=risk_h(ctx.h["advisor"]), params={"accept_date": inside})
        expect_code(G, "advisor 确认 → 403", r, 403, "AUTH_403_AGENT_MISMATCH")



GROUPS = {"R1": r1, "R2": r2, "R3": r3, "R4": r4, "R5": r5, "R6": r6}


# --------------------------------------------------------------------------
# 主流程
# --------------------------------------------------------------------------
def main() -> int:
    parser = argparse.ArgumentParser(description="风控 Agent 端到端 API 冒烟")
    parser.add_argument("--base-url", default="http://127.0.0.1:8000")
    parser.add_argument("--report", default=None, help="可选：把结果矩阵落成 Markdown 文件")
    parser.add_argument("--only", action="append", default=None, help="只跑指定分组（如 --only R6）")
    parser.add_argument("--timeout", type=float, default=90.0)
    args = parser.parse_args()

    try:
        sys.stdout.reconfigure(encoding="utf-8")
    except Exception:
        pass

    ctx = Ctx(client=httpx.Client(base_url=args.base_url.rstrip("/"), timeout=args.timeout))
    try:
        if not preflight(ctx, args.base_url):
            print("\n前置自检未通过 —— 未产生任何用例判定。")
            return 2

        # 顺序有依赖：R5 造预警 → R2 看统计 / R3 处置
        order = ["R1", "R5", "R2", "R3", "R4", "R6"]
        selected = [g for g in order if not args.only or g in args.only]
        print(f"\n-- 执行分组: {', '.join(selected)} --\n")
        for name in selected:
            GROUPS[name](ctx)
    finally:
        ctx.client.close()

    counts = {s: sum(1 for r in RESULTS if r.status == s) for s in (PASS, FAIL, SKIP)}
    print(f"\n== 结果: {counts[PASS]} PASS / {counts[FAIL]} FAIL / {counts[SKIP]} SKIP ==")
    for r in RESULTS:
        if r.status == FAIL:
            print(f"  - {r.group} · {r.name}: 期望 {r.expected} / 实际 {r.actual}")

    if args.report:
        write_report(Path(args.report))
        print(f"\n报告已写入 {args.report}")
    else:
        print("\n（未落盘；如需矩阵加 --report <path>）")

    return 1 if counts[FAIL] else 0


def write_report(path: Path) -> None:
    lines = ["| 分组 | 用例 | 状态 | 期望 | 实际 | 证据 |", "| --- | --- | --- | --- | --- | --- |"]
    for r in RESULTS:
        ev = r.evidence.replace("|", "\\|").replace("\n", " ")[:200]
        lines.append(f"| {r.group} | {r.name} | {r.status} | {r.expected} | {r.actual} | {ev} |")
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text("\n".join(lines) + "\n", encoding="utf-8")


if __name__ == "__main__":
    sys.exit(main())