group_xinghuo_jinrong/tests/test_convert_service.py

"""T-7 `convert_service` 八步编排单测（开发计划 §6.2 DoD）。

sqlite 内存库（`conftest.sqlite_engine`，含 C×R 矩阵种子），数据自建。
折算期望值一律由生产 `calc.py` 实算（不自造公式副本，自检第 13 问）。

覆盖：
1. 三阶段贯通（占位 pending→completed / 2 流水 / 响应字段与 PRD §5.3 对齐）
2. **blocked 不占位**（PRD §7.0：前四步不落库）
3. 幂等命中返回首次结果、**不产生第二组流水**
4. 未抢到执行权 → `status=processing`（T-9 映射 202）
5. 全额转出豁免 `min_redeem_qty`（验收 16）/ 强制全转留痕
6. `nav_stale` 额外落副审计（1~2 条）
7. 阶段 1.5 引擎异常**不阻断**已成立交易
8. 阶段二失败 → `convert_detail_write_failed` 审计 + 占位 `failed`
9. 各 4xx/503 分支：SameProduct / 不可赎回 / 跨主体 / 份额不足 / 低于最低份额 /
   无净值 / 批次数超限
"""

from __future__ import annotations

import json
from datetime import date, datetime, timedelta
from decimal import Decimal

import pytest
from sqlalchemy import text

from app.gateway.convert_core_repository import ConvertCoreRepository
from app.repository.convert_repository import ConvertRepository
from app.repository.core_ro import CoreReadOnlyRepository
from app.repository.risk_repository import RiskRepository
from app.service.convert.calc import (
    convert_amount,
    diff_fee,
    hold_days,
    in_qty,
    lot_amount,
    lot_fee,
    plan_lots,
)
from app.service.convert.convert_service import (
    PROCESSING,
    compensate_convert,
    convert_fund,
)
from app.service.convert.errors import (
    BelowMinQty,
    CrossEntityNotSupported,
    InsufficientShares,
    NavNotReady,
    ProductNotRedeemable,
    SameProduct,
    TooManyLots,
)
from app.service.convert.fee import pick_fee_rate
from app.service.convert.types import FeeRule, Lot
from app.service.risk.locks import try_lock
from app.service.risk.rules import RiskThresholds

CUST = "CUST-T7"  # C3 客户 → R4 产品 allowed_with_disclosure（放行）
CUST_LOW = "CUST-T7L"  # C1 客户 → R4 forbidden（用于 blocked 分支）
PROD_OUT = "PROD-T7A"
PROD_IN = "PROD-T7B"
COMPANY = "华夏模拟基金"
TA = "TA-CN-001"
NOW = datetime(2026, 9, 4, 10, 0, 0)
TODAY = date(2026, 9, 4)
OUT_RATE = Decimal("0.0030")
IN_RATE = Decimal("0.0080")
IN_NAV = Decimal("0.9500")
FEE_TIERS = [
    (0, 7, "0.0150"), (7, 30, "0.0100"), (30, 180, "0.0050"),
    (180, 365, "0.0025"), (365, None, "0.0000"),
]


def _exec(engine, sql: str, **params) -> None:
    with engine.begin() as conn:
        conn.execute(text(sql), params)


def _rows(engine, sql: str, **params) -> list[dict]:
    with engine.connect() as conn:
        return [dict(r) for r in conn.execute(text(sql), params).mappings()]


def _seed(engine, *, nav_date: date = TODAY, min_redeem: str = "0", min_hold: str = "0") -> None:
    _exec(
        engine,
        "INSERT INTO core_customer (customer_id, display_name, age, is_active) "
        "VALUES (:c, 'T7客户', 40, 1)",
        c=CUST,
    )
    _exec(
        engine,
        "INSERT INTO core_customer (customer_id, display_name, age, is_active) "
        "VALUES (:c, 'T7低风险客户', 40, 1)",
        c=CUST_LOW,
    )
    _exec(
        engine,
        "INSERT INTO core_customer_risk (customer_id, risk_code, evaluated_at, expires_at) "
        "VALUES (:c, 'C3', :t, :exp)",
        c=CUST, t=NOW - timedelta(days=30), exp=NOW + timedelta(days=300),
    )
    _exec(
        engine,
        "INSERT INTO core_customer_risk (customer_id, risk_code, evaluated_at, expires_at) "
        "VALUES (:c, 'C1', :t, :exp)",
        c=CUST_LOW, t=NOW - timedelta(days=30), exp=NOW + timedelta(days=300),
    )
    _exec(
        engine,
        "INSERT INTO core_product (product_id, product_name, min_risk_code, product_type, "
        "can_subscribe, can_redeem, subscribe_fee_rate, min_redeem_qty, min_hold_qty, "
        "fund_company, ta_code) VALUES (:p, '转出债基', 'R2', 'bond', 1, 1, :r, :mr, :mh, :co, :ta)",
        p=PROD_OUT, r=float(OUT_RATE), mr=float(min_redeem), mh=float(min_hold),
        co=COMPANY, ta=TA,
    )
    _exec(
        engine,
        "INSERT INTO core_product (product_id, product_name, min_risk_code, product_type, "
        "can_subscribe, can_redeem, subscribe_fee_rate, fund_company, ta_code) "
        "VALUES (:p, '转入股基', 'R4', 'stock', 1, 1, :r, :co, :ta)",
        p=PROD_IN, r=float(IN_RATE), co=COMPANY, ta=TA,
    )
    for mh, mh_max, rate in FEE_TIERS:
        _exec(
            engine,
            "INSERT INTO core_fee_rule (product_id, fee_type, min_hold_days, max_hold_days, rate) "
            "VALUES (:p, 'redeem', :mh, :mh_max, :rate)",
            p=PROD_OUT, mh=mh, mh_max=mh_max, rate=float(rate),
        )
    _exec(
        engine,
        "INSERT INTO core_product_nav (product_id, nav, daily_chg_pct, nav_date) "
        "VALUES (:p, :nav, 0, :d)",
        p=PROD_IN, nav=float(IN_NAV), d=nav_date,
    )
    _seed_lot(engine, "LOT-T7-1", "100", "1.0300", datetime(2026, 8, 1, 10, 0, 0))
    _seed_lot(engine, "LOT-T7-2", "50", "1.0000", datetime(2026, 9, 1, 10, 0, 0))
    _exec(
        engine,
        "INSERT INTO core_holding (customer_id, product_id, qty, cost_amount, market_value, "
        "pnl_pct, as_of) VALUES (:c, :p, 150, 150, 154.5, 0, :d)",
        c=CUST, p=PROD_OUT, d=TODAY,
    )
    # CUST_LOW（C1，R4 应被适当性拦截）也需有份额，否则 ② 份额校验会先于 ④ 触发
    _seed_lot(engine, "LOT-T7L-1", "100", "1.0300", datetime(2026, 8, 1, 10, 0, 0),
              customer=CUST_LOW)
    _seed_lot(engine, "LOT-T7L-2", "50", "1.0000", datetime(2026, 9, 1, 10, 0, 0),
              customer=CUST_LOW)
    _exec(
        engine,
        "INSERT INTO core_holding (customer_id, product_id, qty, cost_amount, market_value, "
        "pnl_pct, as_of) VALUES (:c, :p, 150, 150, 154.5, 0, :d)",
        c=CUST_LOW, p=PROD_OUT, d=TODAY,
    )


def _seed_lot(engine, lot_id: str, qty: str, nav: str, confirmed_at: datetime,
              customer: str = CUST) -> None:
    _exec(
        engine,
        "INSERT INTO core_share_lot (lot_id, customer_id, product_id, qty, remain_qty, nav, "
        "confirmed_at) VALUES (:l, :c, :p, :q, :q, :nav, :cat)",
        l=lot_id, c=customer, p=PROD_OUT, q=float(qty), nav=float(nav), cat=confirmed_at,
    )


def _services(engine):
    return dict(
        core_ro=CoreReadOnlyRepository(engine=engine),
        risk_repo=RiskRepository(engine=engine),
        convert_repo=ConvertRepository(engine=engine),
        core_writer=ConvertCoreRepository(engine=engine),
    )


def _req(customer: str = CUST, qty: str = "120", cid_req: str | None = None) -> dict:
    return {
        "customer_id": customer,
        "from_product_id": PROD_OUT,
        "to_product_id": PROD_IN,
        "qty": qty,
        "client_request_id": cid_req,
    }


def _expected(engine, requested: str) -> dict:
    """用生产纯函数算一遍期望值（与 service 内部同一套口径）。"""
    core = CoreReadOnlyRepository(engine=engine)
    lots = [Lot.from_row(r) for r in core.list_share_lots(CUST, PROD_OUT)]
    rules = [FeeRule.from_row(r) for r in core.get_redeem_fee_rules(PROD_OUT)]
    plan = plan_lots(lots, Decimal(requested))
    out_amount = Decimal("0")
    redeem_fee = Decimal("0")
    for alloc in plan.allocations:
        amount = lot_amount(alloc.qty, alloc.nav)
        rate = pick_fee_rate(rules, hold_days(TODAY, alloc.confirmed_at), product_id=PROD_OUT)
        out_amount += amount
        redeem_fee += lot_fee(amount, rate)
    conv = convert_amount(out_amount, redeem_fee)
    gap = diff_fee(conv, OUT_RATE, IN_RATE, "amount_diff")
    in_amount = conv - gap
    return {
        "out_amount": out_amount,
        "redeem_fee": redeem_fee,
        "convert_amount": conv,
        "diff_fee": gap,
        "in_amount": in_amount,
        "in_qty": in_qty(in_amount, IN_NAV),
        "actual_qty": plan.actual_qty,
    }


# ── 1. 三阶段贯通 ───────────────────────────────────────────────────
def test_happy_path_writes_two_trades_and_completes_placeholder(sqlite_engine):
    _seed(sqlite_engine)
    exp = _expected(sqlite_engine, "120")
    resp = convert_fund(_req(), now=NOW, **_services(sqlite_engine))

    assert resp["blocked"] is False
    assert resp["estimated"] is True
    assert resp["convert_group_id"].startswith("CNV-")
    assert Decimal(resp["out_amount"]) == exp["out_amount"]
    assert Decimal(resp["redeem_fee"]) == exp["redeem_fee"]
    assert Decimal(resp["convert_amount"]) == exp["convert_amount"]
    assert Decimal(resp["diff_fee"]) == exp["diff_fee"]
    assert Decimal(resp["in_amount"]) == exp["in_amount"]
    assert Decimal(resp["in_qty"]) == exp["in_qty"]
    assert Decimal(resp["actual_qty"]) == exp["actual_qty"]
    assert resp["lot_count"] == 2
    assert [b["hold_days"] for b in resp["lot_breakdown"]] == [34, 3]
    assert resp["confirm_basis"] == "natural_day_approx"

    # 阶段二：占位 completed
    rows = _rows(
        sqlite_engine,
        "SELECT * FROM risk_convert_detail WHERE convert_group_id = :g",
        g=resp["convert_group_id"],
    )
    assert len(rows) == 1 and rows[0]["status"] == "completed"
    # 两条流水同组、redeem/subscribe（R-b）
    trades = _rows(
        sqlite_engine,
        "SELECT * FROM core_trade WHERE convert_group_id = :g ORDER BY trade_type",
        g=resp["convert_group_id"],
    )
    assert [t["trade_type"] for t in trades] == ["redeem", "subscribe"]
    # 主审计 1 条（净值新鲜 → 无 nav_stale 副审计）
    assert len(
        _rows(sqlite_engine, "SELECT * FROM audit_log WHERE decision = 'convert_accepted'")
    ) == 1
    assert _rows(sqlite_engine, "SELECT * FROM audit_log WHERE decision = 'nav_stale'") == []


# ── 2. blocked 不占位（PRD §7.0）─────────────────────────────────────
def test_suitability_blocked_does_not_placeholder(sqlite_engine):
    _seed(sqlite_engine)
    resp = convert_fund(
        _req(customer=CUST_LOW), now=NOW, **_services(sqlite_engine)
    )
    assert resp["blocked"] is True
    assert resp["block_response_code"]
    # 关键：前四步不落库 —— 无占位、无流水
    assert _rows(sqlite_engine, "SELECT * FROM risk_convert_detail") == []
    assert _rows(sqlite_engine, "SELECT * FROM core_trade") == []
    assert len(
        _rows(sqlite_engine, "SELECT * FROM audit_log WHERE decision = 'suitability_blocked'")
    ) == 1


# ── 3. 幂等：同键重复提交不产生第二组流水 ────────────────────────────
def test_idempotent_repeat_returns_first_result(sqlite_engine):
    _seed(sqlite_engine)
    svc = _services(sqlite_engine)
    first = convert_fund(_req(cid_req="CLI-T7-001"), now=NOW, **svc)
    again = convert_fund(_req(cid_req="CLI-T7-001"), now=NOW, **svc)

    assert again["convert_group_id"] == first["convert_group_id"]
    assert again["in_qty"] == first["in_qty"]
    assert again["out_trade_id"] == first["out_trade_id"]
    # 只有一组流水（2 条），没有第二组
    assert len(_rows(sqlite_engine, "SELECT * FROM core_trade")) == 2
    assert len(_rows(sqlite_engine, "SELECT * FROM risk_convert_detail")) == 1


# ── 4. 未抢到执行权 → processing（T-9 映射 202）───────────────────────
def test_lock_not_acquired_returns_processing(sqlite_engine, monkeypatch):
    import app.service.convert.convert_service as cs
    from app.service.risk.locks import _NoLock

    _seed(sqlite_engine)
    monkeypatch.setattr(cs, "try_lock", lambda *a, **k: _NoLock())
    resp = convert_fund(_req(cid_req="CLI-T7-002"), now=NOW, **_services(sqlite_engine))

    assert resp["status"] == PROCESSING
    assert "convert_group_id" in resp
    assert _rows(sqlite_engine, "SELECT * FROM core_trade") == []
    assert _rows(sqlite_engine, "SELECT * FROM risk_convert_detail") == []


# ── 5. 全额转出豁免最低份额（验收 16）────────────────────────────────
def test_full_transfer_waives_min_redeem_qty(sqlite_engine):
    _seed(sqlite_engine, min_redeem="10000")
    # 持 150（< min_redeem 10000）但申请全额 → 豁免，成功
    resp = convert_fund(_req(qty="150"), now=NOW, **_services(sqlite_engine))
    assert resp["blocked"] is False
    assert Decimal(resp["actual_qty"]) == Decimal("150")


def test_below_min_qty_rejected_when_not_full(sqlite_engine):
    _seed(sqlite_engine, min_redeem="10000")
    # 持 150、申请 100（< 最低 10000，且非全额）→ 400 BELOW_MIN_QTY
    with pytest.raises(BelowMinQty):
        convert_fund(_req(qty="100"), now=NOW, **_services(sqlite_engine))


# ── 6. 强制全转留痕 ─────────────────────────────────────────────────
def test_forced_full_transfer_flag(sqlite_engine):
    _seed(sqlite_engine, min_hold="100")
    # 持 150、申请 140 → 余额 10 < 100 → 强制全转 150
    resp = convert_fund(_req(qty="140"), now=NOW, **_services(sqlite_engine))
    assert resp["forced_full_transfer"] is True
    assert resp["min_hold_action"] == "force_transfer"
    assert Decimal(resp["actual_qty"]) == Decimal("150")


# ── 7. nav_stale 额外落副审计 ────────────────────────────────────────
def test_nav_stale_adds_second_audit(sqlite_engine):
    _seed(sqlite_engine, nav_date=date(2026, 8, 25))  # 距今 10 天 > 3
    resp = convert_fund(_req(), now=NOW, **_services(sqlite_engine))
    assert resp["nav_stale"] is True
    assert len(_rows(sqlite_engine, "SELECT * FROM audit_log WHERE decision = 'nav_stale'")) == 1
    assert len(
        _rows(sqlite_engine, "SELECT * FROM audit_log WHERE decision = 'convert_accepted'")
    ) == 1


# ── 8. 阶段 1.5：引擎异常不阻断已成立的交易 ───────────────────────────
def test_engine_exception_does_not_block_trade(sqlite_engine):
    _seed(sqlite_engine)

    def boom(out_trade, in_trade):  # noqa: ANN001
        raise RuntimeError("引擎炸了")

    resp = convert_fund(_req(), now=NOW, engine_hook=boom, **_services(sqlite_engine))
    assert resp["blocked"] is False
    assert resp["engine_error"] is True
    assert len(_rows(sqlite_engine, "SELECT * FROM core_trade")) == 2  # 交易仍成立
    assert len(_rows(sqlite_engine, "SELECT * FROM audit_log WHERE decision = 'engine_error'")) == 1


def test_engine_result_merged_into_response(sqlite_engine):
    _seed(sqlite_engine)
    hook = lambda o, i: {  # noqa: E731
        "triggered_rules": ["RISK-002"],
        "alert_ids": ["ALT-1"],
        "aml_hit": False,
    }
    resp = convert_fund(_req(), now=NOW, engine_hook=hook, **_services(sqlite_engine))
    assert resp["triggered_rules"] == ["RISK-002"]
    assert resp["alert_ids"] == ["ALT-1"]


# ── 9. 阶段二失败：留痕 + 占位 failed（不回滚 Core）───────────────────
def test_phase_two_failure_keeps_trade_and_marks_failed(sqlite_engine, monkeypatch):
    _seed(sqlite_engine)

    def boom(self, group_id, **kwargs):  # noqa: ANN001
        raise RuntimeError("阶段二写失败")

    monkeypatch.setattr(ConvertRepository, "complete_convert", boom)
    resp = convert_fund(
        _req(cid_req="CLI-T7-003"), now=NOW, **_services(sqlite_engine)
    )
    # 交易已成立：core 有两条流水，响应照常返回
    assert len(_rows(sqlite_engine, "SELECT * FROM core_trade")) == 2
    assert resp["blocked"] is False
    assert len(
        _rows(
            sqlite_engine,
            "SELECT * FROM audit_log WHERE decision = 'convert_detail_write_failed'",
        )
    ) == 1
    rows = _rows(sqlite_engine, "SELECT * FROM risk_convert_detail")
    assert rows[0]["status"] == "failed"


# ── 10. 各 4xx / 503 分支 ───────────────────────────────────────────
def test_same_product_rejected(sqlite_engine):
    _seed(sqlite_engine)
    req = _req()
    req["to_product_id"] = PROD_OUT
    with pytest.raises(SameProduct):
        convert_fund(req, now=NOW, **_services(sqlite_engine))


def test_cross_entity_rejected(sqlite_engine):
    _seed(sqlite_engine)
    _exec(
        sqlite_engine,
        "UPDATE core_product SET fund_company = '易方达模拟基金' WHERE product_id = :p",
        p=PROD_IN,
    )
    with pytest.raises(CrossEntityNotSupported):
        convert_fund(_req(), now=NOW, **_services(sqlite_engine))


def test_out_product_not_redeemable_rejected(sqlite_engine):
    _seed(sqlite_engine)
    _exec(
        sqlite_engine,
        "UPDATE core_product SET can_redeem = 0 WHERE product_id = :p",
        p=PROD_OUT,
    )
    with pytest.raises(ProductNotRedeemable):
        convert_fund(_req(), now=NOW, **_services(sqlite_engine))


def test_insufficient_shares_rejected(sqlite_engine):
    _seed(sqlite_engine)
    with pytest.raises(InsufficientShares):
        convert_fund(_req(qty="500"), now=NOW, **_services(sqlite_engine))


def test_nav_not_ready_503(sqlite_engine):
    _seed(sqlite_engine)
    _exec(sqlite_engine, "DELETE FROM core_product_nav")
    with pytest.raises(NavNotReady) as exc:
        convert_fund(_req(), now=NOW, **_services(sqlite_engine))
    assert exc.value.status_code == 503


def test_too_many_lots_rejected(sqlite_engine, monkeypatch):
    from app.config import settings as settings_module

    _seed(sqlite_engine)
    monkeypatch.setattr(settings_module.settings, "convert_batch_max_lots", 1)
    with pytest.raises(TooManyLots) as exc:
        convert_fund(_req(qty="120"), now=NOW, **_services(sqlite_engine))
    assert exc.value.extra == {"batch_count": 2, "max_lots": 1}


# ── 10. 阶段 1.5 接线（T-8）：引擎真跑并出单 ─────────────────────────
def test_engine_wired_produces_single_alert_with_two_events(sqlite_engine):
    """T-8 落地后阶段 1.5 不再跳过：大额转换**真出一张单**、`payload.events` 两条（验收 7）。

    本用例是「接线回归」：若 `_run_engine` 又被改回静默跳过（或签名对不上被
    ImportError 吞掉），这里会因 `triggered_rules` 为空而变红。
    """
    _seed(sqlite_engine)
    th = RiskThresholds(
        large_amount=Decimal("100"),  # 调低以让 120 份的折算额命中 RISK-001
        daily_total=Decimal("1000000"),
        freq_count=3,
        probe_window_minutes=5,
        probe_count=3,
        probe_amount=Decimal("400000"),
        small_amount=Decimal("10000"),
        small_count=3,
        concentration_threshold=1.01,
    )
    resp = convert_fund(_req(), now=NOW, thresholds=th, **_services(sqlite_engine))

    assert resp["engine_error"] is False, "引擎真的跑了且没炸（未走 ImportError 跳过分支）"
    assert "RISK-001" in resp["triggered_rules"]
    assert len(resp["alert_ids"]) == 1, "一次转换只出一张单"

    alerts = _rows(sqlite_engine, "SELECT * FROM risk_alert")
    assert len(alerts) == 1
    payload = json.loads(alerts[0]["payload"])
    assert len(payload["events"]) == 2, "一张单承载两条事件（转出 + 转入）"
    assert [e["trade_type"] for e in payload["events"]] == ["redeem", "subscribe"]


# ── 11. T-12 补偿：阶段二失败 → 按 group 补写详情 + 预警（FR-C17 / 架构 §5.4）──
def _low_thresholds() -> RiskThresholds:
    """调低大额阈值，让 120 份的折算额足以命中 RISK-001（与 §10 接线用例同口径）。"""
    return RiskThresholds(
        large_amount=Decimal("100"),
        daily_total=Decimal("1000000"),
        freq_count=3,
        probe_window_minutes=5,
        probe_count=3,
        probe_amount=Decimal("400000"),
        small_amount=Decimal("10000"),
        small_count=3,
        concentration_threshold=1.01,
    )


def _prepare_failed_convert(sqlite_engine, monkeypatch, cid_req: str) -> str:
    """构造**真实待补偿态**：Core 侧两条流水已成立，但详情与预警双双缺失。

    - 阶段二：`complete_convert` 打桩抛异常 → 占位 `failed` + `convert_detail_write_failed` 审计；
    - 阶段 1.5：注入会抛的 `engine_hook` → 无预警单 + `engine_error` 审计。

    这正是 PRD §7.1 描述的补偿场景（agent 侧两件事都没落）。
    返回 convert_group_id；`complete_convert` 已被还原（补偿才能真的写进去）。
    """
    _seed(sqlite_engine)
    original = ConvertRepository.complete_convert

    def boom(self, group_id, **kwargs):  # noqa: ANN001
        raise RuntimeError("阶段二写失败")

    def engine_boom(out_trade, in_trade):  # noqa: ANN001
        raise RuntimeError("阶段 1.5 引擎失败")

    monkeypatch.setattr(ConvertRepository, "complete_convert", boom)
    resp = convert_fund(
        _req(cid_req=cid_req),
        now=NOW,
        engine_hook=engine_boom,
        thresholds=_low_thresholds(),
        **_services(sqlite_engine),
    )
    monkeypatch.setattr(ConvertRepository, "complete_convert", original)
    return resp["convert_group_id"]


def _compensate(sqlite_engine, gid: str) -> dict:
    svc = _services(sqlite_engine)
    return compensate_convert(
        gid,
        core_ro=svc["core_ro"],
        risk_repo=svc["risk_repo"],
        convert_repo=svc["convert_repo"],
        thresholds=_low_thresholds(),
        now=NOW,
    )


def _detail_rows(engine, gid: str) -> list[dict]:
    return _rows(
        engine, "SELECT * FROM risk_convert_detail WHERE convert_group_id = :g", g=gid
    )


def test_compensate_rebuilds_detail_and_alerts(sqlite_engine, monkeypatch):
    """补偿把「详情 + 预警」两件事都补齐，且详情直接置 completed（验收 17）。"""
    gid = _prepare_failed_convert(sqlite_engine, monkeypatch, "CLI-T12-001")
    # 前置态自检：交易已成立、详情 failed、无预警单
    assert len(_rows(sqlite_engine, "SELECT * FROM core_trade")) == 2
    assert _detail_rows(sqlite_engine, gid)[0]["status"] == "failed"
    assert _rows(sqlite_engine, "SELECT * FROM risk_alert") == []
    assert len(
        _rows(
            sqlite_engine,
            "SELECT * FROM audit_log WHERE decision = 'convert_detail_write_failed'",
        )
    ) == 1

    out = _compensate(sqlite_engine, gid)

    assert out["state"] == "rebuilt"
    assert out["detail"] == "rebuilt" and out["engine"] == "rebuilt"
    assert "RISK-001" in out["triggered_rules"]
    assert len(out["alert_ids"]) == 1
    row = _detail_rows(sqlite_engine, gid)[0]
    assert row["status"] == "completed"
    # 补偿回填的指针与 Core 侧一致（幂等锚点 = 转出端）
    assert row["out_trade_id"] == out["out_trade_id"]
    assert row["in_trade_id"] == out["in_trade_id"]
    assert row["nav"] is not None and row["nav_date"] is not None
    assert len(_rows(sqlite_engine, "SELECT * FROM risk_alert")) == 1
    # 补偿也补了主审计（与首次成功路径同一份 `_write_main_audit`）
    assert len(
        _rows(
            sqlite_engine,
            "SELECT * FROM audit_log WHERE decision = 'convert_accepted'",
        )
    ) == 1


def test_compensate_is_idempotent(sqlite_engine, monkeypatch):
    """重复补偿 → `skipped`，不产生第二张预警单、不重复写详情与主审计。"""
    gid = _prepare_failed_convert(sqlite_engine, monkeypatch, "CLI-T12-002")
    first = _compensate(sqlite_engine, gid)
    second = _compensate(sqlite_engine, gid)

    assert first["state"] == "rebuilt"
    assert second["state"] == "skipped"
    assert second["detail"] == "already_completed"
    assert second["engine"] == "skipped"
    assert second["alert_ids"] == first["alert_ids"]
    assert second["triggered_rules"] == []
    assert len(_rows(sqlite_engine, "SELECT * FROM risk_alert")) == 1  # 只有一张单
    assert len(
        _rows(
            sqlite_engine,
            "SELECT * FROM audit_log WHERE decision = 'convert_accepted'",
        )
    ) == 1  # 主审计不重复
    # 曾阶段 1.5 中断过 → 幂等跳过时附「人工核对」提示（与 rebuild_alerts 同款警示）
    assert "人工核对" in second["warning"]


def test_compensate_missing_group_touches_nothing(sqlite_engine):
    """Core 侧不足两条流水 → missing，零写入（不是有效转换组）。"""
    _seed(sqlite_engine)
    out = _compensate(sqlite_engine, "CNV-T12-NOPE")
    assert out["state"] == "missing"
    assert out["trade_count"] == 0
    assert _rows(sqlite_engine, "SELECT * FROM risk_alert") == []
    assert _rows(sqlite_engine, "SELECT * FROM audit_log") == []
    assert _rows(sqlite_engine, "SELECT * FROM risk_convert_detail") == []


def test_compensate_returns_locked_when_execution_right_taken(
    sqlite_engine, monkeypatch
):
    """未抢到 `convert:rerun:{gid}` → locked（有并发重试/实例在跑），零写入。"""
    gid = _prepare_failed_convert(sqlite_engine, monkeypatch, "CLI-T12-004")
    with try_lock(f"convert:rerun:{gid}", 30) as acquired:
        assert acquired is True
        out = _compensate(sqlite_engine, gid)
    assert out["state"] == "locked" and out["alert_ids"] == []
    assert _detail_rows(sqlite_engine, gid)[0]["status"] == "failed"  # 仍待补偿
    assert _rows(sqlite_engine, "SELECT * FROM risk_alert") == []