X-Financial/server/tests/test_risk_dispositions.py

from __future__ import annotations

from collections.abc import Generator
from datetime import UTC, datetime, timedelta
from decimal import Decimal

import pytest
from auth_helpers import install_legacy_header_auth_override
from fastapi import FastAPI
from fastapi.testclient import TestClient
from sqlalchemy import create_engine, func, select
from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy.pool import StaticPool

from app.api.deps import CurrentUserContext, get_db
from app.api.v1.endpoints.risk_observations import router as risk_observations_router
from app.core.agent_enums import AgentAssetDomain, AgentAssetStatus, AgentAssetType
from app.db.base import Base
from app.models.agent_asset import AgentAsset
from app.models.agent_asset_release_telemetry import AgentAssetReleaseLabel
from app.models.employee import Employee
from app.models.financial_record import ExpenseClaim
from app.models.risk_disposition import RiskDispositionEvent
from app.models.risk_observation import RiskObservationFeedback
from app.schemas.risk_disposition import RiskDispositionActionCreate
from app.services.agent_asset_release_telemetry import (
    AgentAssetReleaseTelemetryService,
    ReleaseObservationInput,
)
from app.services.risk_dispositions import (
    RiskDispositionConflictError,
    RiskDispositionIdempotencyConflictError,
    RiskDispositionPermissionError,
    RiskDispositionService,
    RiskDispositionVersionConflictError,
    RiskWaiverDecisionPermissionError,
)
from app.services.risk_observation_access_policy import RiskObservationAccessPolicy
from app.services.risk_observations import RiskObservationService


def test_risk_disposition_separates_adjudication_and_lifecycle_with_audit_events(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setenv("FEW_SHOT_INJECTION_ENABLED", "false")
    with _build_session() as db:
        observation = RiskObservationService(db).upsert_observation(
            _observation_payload("risk:typed:duplicate")
        )
        db.commit()
        service = RiskDispositionService(db)

        confirmed = service.execute_action(
            observation.id,
            _action("confirm", version=0, request_id="request-confirm-001"),
            tenant_id="default",
            actor_id="finance-1",
            actor_name="财务甲",
        )
        assert confirmed.disposition.adjudication == "confirmed"
        assert confirmed.disposition.lifecycle_status == "open"
        assert confirmed.disposition.version == 1
        assert confirmed.event.before_json["adjudication"] == "unreviewed"
        assert confirmed.event.after_json["adjudication"] == "confirmed"

        supplemented = service.execute_action(
            observation.id,
            RiskDispositionActionCreate(
                action="request_supplement",
                expected_version=1,
                request_id="request-supplement-001",
                assignee="员工甲",
                due_at=datetime.now(UTC) + timedelta(days=2),
                comment="请补齐行程单",
            ),
            tenant_id="default",
            actor_id="finance-1",
            actor_name="财务甲",
        )
        assert supplemented.disposition.adjudication == "confirmed"
        assert supplemented.disposition.lifecycle_status == "supplement_requested"
        assert supplemented.disposition.assignee == "员工甲"
        assert supplemented.disposition.version == 2

        resolved = service.execute_action(
            observation.id,
            RiskDispositionActionCreate(
                action="resolve",
                expected_version=2,
                request_id="request-resolve-001",
                resolution="补充材料已核验，风险关闭。",
            ),
            tenant_id="default",
            actor_id="finance-1",
            actor_name="财务甲",
        )
        assert resolved.disposition.adjudication == "confirmed"
        assert resolved.disposition.lifecycle_status == "resolved"
        assert resolved.disposition.resolution == "补充材料已核验，风险关闭。"
        assert resolved.disposition.version == 3
        assert db.scalar(select(func.count()).select_from(RiskDispositionEvent)) == 3
        assert db.scalar(select(func.count()).select_from(RiskObservationFeedback)) == 1

        with pytest.raises(RiskDispositionConflictError, match="已解决"):
            service.execute_action(
                observation.id,
                _action(
                    "start_remediation",
                    version=3,
                    request_id="request-remediation-after-resolve",
                ),
                tenant_id="default",
                actor_id="finance-1",
                actor_name="财务甲",
            )
        with pytest.raises(RiskDispositionConflictError, match="已解决"):
            service.execute_action(
                observation.id,
                _action(
                    "false_positive",
                    version=3,
                    request_id="request-readjudicate-after-resolve",
                ),
                tenant_id="default",
                actor_id="finance-1",
                actor_name="财务甲",
            )


def test_typed_risk_disposition_appends_current_release_label_automatically(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setenv("FEW_SHOT_INJECTION_ENABLED", "false")
    with _build_session() as db:
        db.add(_claim())
        asset = AgentAsset(
            id="risk-disposition-release-asset",
            tenant_id="default",
            scope="tenant",
            asset_type=AgentAssetType.RULE.value,
            code="risk.disposition.release",
            name="处置学习发布规则",
            description="",
            domain=AgentAssetDomain.EXPENSE.value,
            scenario_json=["travel"],
            owner="finance",
            status=AgentAssetStatus.ACTIVE.value,
            current_version="v1",
            published_version="v1",
            working_version="v2",
            config_json={
                "tenant_id": "default",
                "detail_mode": "json_risk",
                "stable_marker": "stable-risk-disposition-release",
                "release_guard": {
                    "release_id": "risk-disposition-release-1",
                    "stage": "shadow",
                    "candidate_version": "v2",
                    "previous_version": "v1",
                    "previous_config": {
                        "tenant_id": "default",
                        "detail_mode": "json_risk",
                        "enabled": True,
                        "stable_marker": "stable-risk-disposition-release",
                    },
                    "policy": {
                        "shadow_min_samples": 1,
                        "canary_min_samples": 1,
                        "max_error_rate": 1.0,
                        "min_precision": 0.9,
                        "max_precision_drop": 1.0,
                        "canary_traffic_percent": 10,
                    },
                },
            },
        )
        db.add(asset)
        db.commit()
        telemetry = AgentAssetReleaseTelemetryService(db).record_observation(
            ReleaseObservationInput(
                tenant_id="default",
                asset_id=asset.id,
                release_id="risk-disposition-release-1",
                stage="shadow",
                version="v2",
                rule_code=asset.code,
                source_key="claim-risk-1",
                candidate_hit=True,
                baseline_hit=True,
                business_stage="reimbursement",
            )
        )
        observation = RiskObservationService(db).upsert_observation(
            {
                **_observation_payload("risk:typed:release-telemetry"),
                "risk_signal": asset.code,
                "algorithm_version": "v1",
                "policy_refs": [asset.code],
                "decision_trace": {"rule_code": asset.code},
            }
        )
        db.commit()

        mutation = RiskDispositionService(db).execute_action(
            observation.id,
            _action("confirm", version=0, request_id="request-release-label-001"),
            tenant_id="default",
            actor_id="finance-secret-account",
            actor_name="财务甲",
        )

        label = db.scalar(
            select(AgentAssetReleaseLabel).where(
                AgentAssetReleaseLabel.observation_id == telemetry.id
            )
        )
        assert mutation.disposition.adjudication == "confirmed"
        assert label is not None
        assert label.label == "confirmed"
        assert label.verification_source == "typed_risk_disposition"
        assert "finance-secret-account" not in str(label.__dict__)

        corrected = RiskDispositionService(db).execute_action(
            observation.id,
            _action(
                "false_positive",
                version=1,
                request_id="request-release-label-correction-001",
            ),
            tenant_id="default",
            actor_id="finance-secret-account",
            actor_name="财务甲",
        )
        db.expire_all()
        refreshed_asset = db.get(AgentAsset, asset.id)

        assert corrected.disposition.adjudication == "false_positive"
        assert refreshed_asset is not None
        assert refreshed_asset.published_version == "v1"
        assert refreshed_asset.config_json["stable_marker"] == ("stable-risk-disposition-release")
        assert refreshed_asset.config_json["release_guard"]["stage"] == "rolled_back"
        assert refreshed_asset.config_json["release_guard"]["rollback"]["automatic"] is True


def test_risk_disposition_requires_confirmation_before_remediation_or_resolution(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setenv("FEW_SHOT_INJECTION_ENABLED", "false")
    with _build_session() as db:
        observation = RiskObservationService(db).upsert_observation(
            _observation_payload("risk:typed:transition-guard")
        )
        db.commit()
        service = RiskDispositionService(db)

        with pytest.raises(RiskDispositionConflictError, match="必须先确认成立"):
            service.execute_action(
                observation.id,
                RiskDispositionActionCreate(
                    action="resolve",
                    expected_version=0,
                    request_id="request-resolve-unreviewed",
                    resolution="不能跳过裁决直接关闭。",
                ),
                tenant_id="default",
                actor_id="finance-1",
                actor_name="财务甲",
            )

        supplemented = service.execute_action(
            observation.id,
            RiskDispositionActionCreate(
                action="request_supplement",
                expected_version=0,
                request_id="request-supplement-unreviewed",
                comment="先补充材料再裁决。",
            ),
            tenant_id="default",
            actor_id="finance-1",
            actor_name="财务甲",
        )
        assert supplemented.disposition.adjudication == "unreviewed"
        assert supplemented.disposition.lifecycle_status == "supplement_requested"


def test_risk_disposition_imports_legacy_confirmation_before_first_typed_action(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setenv("FEW_SHOT_INJECTION_ENABLED", "false")
    with _build_session() as db:
        observation = RiskObservationService(db).upsert_observation(
            _observation_payload("risk:typed:legacy-confirmed")
        )
        observation.status = "confirmed"
        observation.feedback_status = "confirmed"
        db.commit()

        resolved = RiskDispositionService(db).execute_action(
            observation.id,
            RiskDispositionActionCreate(
                action="resolve",
                expected_version=0,
                request_id="request-resolve-legacy-confirmed",
                resolution="历史确认风险已完成整改。",
            ),
            tenant_id="default",
            actor_id="finance-1",
            actor_name="财务甲",
        )

        assert resolved.event.before_json["adjudication"] == "confirmed"
        assert resolved.disposition.adjudication == "confirmed"
        assert resolved.disposition.lifecycle_status == "resolved"


def test_risk_disposition_idempotency_and_optimistic_version_are_enforced(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setenv("FEW_SHOT_INJECTION_ENABLED", "false")
    with _build_session() as db:
        observation = RiskObservationService(db).upsert_observation(
            _observation_payload("risk:typed:idempotency")
        )
        second_observation = RiskObservationService(db).upsert_observation(
            {
                **_observation_payload("risk:typed:idempotency:second"),
                "claim_id": None,
                "claim_no": "",
                "subject_key": "standalone:second",
            }
        )
        db.commit()
        service = RiskDispositionService(db)
        payload = _action("confirm", version=0, request_id="request-idempotent-001")

        first = service.execute_action(
            observation.id,
            payload,
            tenant_id="default",
            actor_id="finance-1",
            actor_name="财务甲",
        )
        replay = service.execute_action(
            observation.id,
            payload,
            tenant_id="default",
            actor_id="finance-1",
            actor_name="财务甲",
        )

        assert replay.replayed is True
        assert replay.event.id == first.event.id
        assert db.scalar(select(func.count()).select_from(RiskDispositionEvent)) == 1
        stored_event = db.get(RiskDispositionEvent, first.event.id)
        assert stored_event is not None
        assert stored_event.response_json is not None
        assert stored_event.response_json["replayed"] is False
        assert stored_event.response_json["disposition"]["version"] == 1

        advanced = service.execute_action(
            observation.id,
            _action(
                "start_remediation",
                version=1,
                request_id="request-remediation-after-idempotent",
            ),
            tenant_id="default",
            actor_id="finance-1",
            actor_name="财务甲",
        )
        late_replay = service.execute_action(
            observation.id,
            payload,
            tenant_id="default",
            actor_id="finance-1",
            actor_name="财务甲",
        )

        assert advanced.disposition.version == 2
        assert late_replay.replayed is True
        assert late_replay.disposition.version == 1
        assert late_replay.disposition.lifecycle_status == "open"
        assert [item.version for item in late_replay.disposition.events] == [1]
        assert db.scalar(select(func.count()).select_from(RiskDispositionEvent)) == 2

        with pytest.raises(RiskDispositionIdempotencyConflictError):
            service.execute_action(
                observation.id,
                _action(
                    "false_positive",
                    version=0,
                    request_id="request-idempotent-001",
                ),
                tenant_id="default",
                actor_id="finance-1",
                actor_name="财务甲",
            )

        with pytest.raises(RiskDispositionIdempotencyConflictError):
            service.execute_action(
                second_observation.id,
                payload,
                tenant_id="default",
                actor_id="finance-1",
                actor_name="财务甲",
            )

        with pytest.raises(RiskDispositionIdempotencyConflictError):
            service.execute_action(
                observation.id,
                payload,
                tenant_id="default",
                actor_id="finance-2",
                actor_name="财务乙",
            )

        with pytest.raises(RiskDispositionVersionConflictError) as error:
            service.execute_action(
                observation.id,
                _action("request_waiver", version=0, request_id="request-stale-001"),
                tenant_id="default",
                actor_id="finance-1",
                actor_name="财务甲",
            )
        assert error.value.current_version == 2


def test_legacy_event_replay_rebuilds_only_target_version_history(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setenv("FEW_SHOT_INJECTION_ENABLED", "false")
    with _build_session() as db:
        observation = RiskObservationService(db).upsert_observation(
            _observation_payload("risk:typed:legacy-replay")
        )
        db.commit()
        service = RiskDispositionService(db)
        payload = _action("confirm", version=0, request_id="request-legacy-replay-001")
        first = service.execute_action(
            observation.id,
            payload,
            tenant_id="default",
            actor_id="finance-1",
            actor_name="财务甲",
        )
        legacy_event = db.get(RiskDispositionEvent, first.event.id)
        assert legacy_event is not None
        legacy_event.response_json = None
        db.commit()

        service.execute_action(
            observation.id,
            _action(
                "start_remediation",
                version=1,
                request_id="request-legacy-replay-later",
            ),
            tenant_id="default",
            actor_id="finance-1",
            actor_name="财务甲",
        )
        replay = service.execute_action(
            observation.id,
            payload,
            tenant_id="default",
            actor_id="finance-1",
            actor_name="财务甲",
        )

        assert replay.replayed is True
        assert replay.event.id == first.event.id
        assert replay.disposition.version == 1
        assert replay.disposition.lifecycle_status == "open"
        assert [item.version for item in replay.disposition.events] == [1]


def test_risk_observation_api_enforces_pool_claim_and_typed_action_permissions(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setenv("FEW_SHOT_INJECTION_ENABLED", "false")
    client, session_factory = _build_client()
    with session_factory() as db:
        db.add(_employee())
        db.add(_claim())
        db.flush()
        observation = RiskObservationService(db).upsert_observation(
            _observation_payload("risk:api:duplicate")
        )
        standalone_observation = RiskObservationService(db).upsert_observation(
            {
                **_observation_payload("risk:api:standalone"),
                "claim_id": None,
                "claim_no": "",
                "subject_key": "default:standalone",
            }
        )
        foreign_observation = RiskObservationService(db).upsert_observation(
            {
                **_observation_payload("risk:api:foreign"),
                "claim_id": None,
                "claim_no": "",
                "subject_key": "tenant-b:standalone",
            },
            tenant_id="tenant-b",
        )
        observation_id = observation.id
        standalone_observation_id = standalone_observation.id
        foreign_observation_id = foreign_observation.id
        db.commit()

    employee_headers = {
        "X-Auth-Username": "risk.employee@example.com",
        "X-Auth-Name": "Risk Employee",
        "X-Auth-Employee-No": "E-RISK",
    }
    finance_headers = {
        "X-Auth-Username": "finance@example.com",
        "X-Auth-Name": "Finance Reviewer",
        "X-Auth-Role-Codes": "finance",
    }

    assert client.get("/api/v1/risk-observations", headers=employee_headers).status_code == 403
    assert (
        client.get(
            f"/api/v1/risk-observations/{observation_id}",
            headers=employee_headers,
        ).status_code
        == 403
    )
    assert (
        client.get(
            "/api/v1/risk-observations/claim/claim-risk-1",
            headers=employee_headers,
        ).status_code
        == 404
    )
    assert (
        client.post(
            f"/api/v1/risk-observations/{observation_id}/feedback",
            headers=employee_headers,
            json={"feedback_type": "confirm"},
        ).status_code
        == 403
    )

    first = client.post(
        f"/api/v1/risk-observations/{observation_id}/disposition/actions",
        headers=finance_headers,
        json={
            "action": "confirm",
            "expected_version": 0,
            "request_id": "api-confirm-request-001",
            "comment": "人工复核确认",
        },
    )
    replay = client.post(
        f"/api/v1/risk-observations/{observation_id}/disposition/actions",
        headers=finance_headers,
        json={
            "action": "confirm",
            "expected_version": 0,
            "request_id": "api-confirm-request-001",
            "comment": "人工复核确认",
        },
    )
    advanced = client.post(
        f"/api/v1/risk-observations/{observation_id}/disposition/actions",
        headers=finance_headers,
        json={
            "action": "request_supplement",
            "expected_version": 1,
            "request_id": "api-supplement-request-001",
            "comment": "请补充完整行程材料",
        },
    )
    late_replay = client.post(
        f"/api/v1/risk-observations/{observation_id}/disposition/actions",
        headers=finance_headers,
        json={
            "action": "confirm",
            "expected_version": 0,
            "request_id": "api-confirm-request-001",
            "comment": "人工复核确认",
        },
    )
    stale = client.post(
        f"/api/v1/risk-observations/{observation_id}/disposition/actions",
        headers=finance_headers,
        json={
            "action": "request_waiver",
            "expected_version": 0,
            "request_id": "api-stale-request-001",
            "waiver_reason": "客户现场条件导致合规材料暂缺",
            "waiver_scope": "仅限本次客户拜访的行程材料",
            "waiver_expires_at": (datetime.now(UTC) + timedelta(days=3)).isoformat(),
        },
    )
    changed_replay = client.post(
        f"/api/v1/risk-observations/{observation_id}/disposition/actions",
        headers=finance_headers,
        json={
            "action": "false_positive",
            "expected_version": 0,
            "request_id": "api-confirm-request-001",
            "comment": "经核验属于误报",
        },
    )
    unsafe_legacy = client.post(
        f"/api/v1/risk-observations/{observation_id}/feedback",
        headers=finance_headers,
        json={"feedback_type": "comment", "payload_json": {"arbitrary": True}},
    )
    foreign_action = client.post(
        f"/api/v1/risk-observations/{foreign_observation_id}/disposition/actions",
        headers=finance_headers,
        json={
            "action": "confirm",
            "expected_version": 0,
            "request_id": "api-cross-tenant-001",
        },
    )
    standalone_action = client.post(
        f"/api/v1/risk-observations/{standalone_observation_id}/disposition/actions",
        headers=finance_headers,
        json={
            "action": "confirm",
            "expected_version": 0,
            "request_id": "api-standalone-finance-001",
        },
    )

    assert first.status_code == 200
    assert first.json()["disposition"]["adjudication"] == "confirmed"
    assert first.json()["disposition"]["lifecycle_status"] == "open"
    assert replay.status_code == 200
    assert replay.json()["replayed"] is True
    assert replay.json()["event"]["id"] == first.json()["event"]["id"]
    assert advanced.status_code == 200
    assert advanced.json()["disposition"]["version"] == 2
    assert late_replay.status_code == 200
    assert late_replay.json()["replayed"] is True
    assert late_replay.json()["disposition"]["version"] == 1
    assert late_replay.json()["disposition"]["lifecycle_status"] == "open"
    assert [item["version"] for item in late_replay.json()["disposition"]["events"]] == [1]
    assert stale.status_code == 409
    assert stale.json()["detail"]["code"] == "RISK_DISPOSITION_VERSION_CONFLICT"
    assert stale.json()["detail"]["message"] == "风险处置状态已更新，请刷新证据链后重试。"
    assert changed_replay.status_code == 409
    assert unsafe_legacy.status_code == 410
    assert foreign_action.status_code == 404
    assert standalone_action.status_code == 403

    detail = client.get(
        f"/api/v1/risk-observations/{observation_id}",
        headers=finance_headers,
    )
    assert detail.status_code == 200
    assert detail.json()["disposition"]["version"] == 2
    assert len(detail.json()["disposition"]["events"]) == 2


def test_current_claim_approver_can_manage_disposition_without_pool_access() -> None:
    with _build_session() as db:
        manager = Employee(
            id="manager-risk",
            employee_no="M-RISK",
            name="风险主管",
            email="risk.manager@example.com",
            position="部门经理",
            grade="P8",
        )
        employee = _employee()
        employee.manager_id = manager.id
        claim = _claim()
        claim.approval_stage = "直属领导审批"
        db.add_all([manager, employee, claim])
        db.flush()
        observation = RiskObservationService(db).upsert_observation(
            _observation_payload("risk:approver:duplicate")
        )
        db.commit()
        current_user = CurrentUserContext(
            tenant_id="default",
            username="risk.manager@example.com",
            name="风险主管",
            role_codes=["approver"],
            is_admin=False,
            employee_no="M-RISK",
        )
        policy = RiskObservationAccessPolicy(db)

        assert policy.can_read_tenant_pool(current_user) is False
        assert policy.can_read_claim_risks(claim.id, current_user) is True
        assert policy.can_manage_disposition(observation, current_user) is True

        unrelated_manager = CurrentUserContext(
            tenant_id="default",
            username="unrelated.manager@example.com",
            name="其他经理",
            role_codes=["manager"],
            is_admin=False,
        )
        finance_outside_stage = CurrentUserContext(
            tenant_id="default",
            username="finance@example.com",
            name="财务甲",
            role_codes=["finance"],
            is_admin=False,
        )
        assert policy.can_read_tenant_pool(unrelated_manager) is False
        assert policy.can_read_claim_risks(claim.id, unrelated_manager) is False
        assert policy.can_manage_disposition(observation, unrelated_manager) is False
        assert policy.can_manage_disposition(observation, finance_outside_stage) is False


def test_disposition_rechecks_current_approver_after_claim_stage_changes() -> None:
    with _build_session() as db:
        manager = Employee(
            id="manager-risk-stage-change",
            employee_no="M-RISK-STAGE",
            name="原审批主管",
            email="risk.stage.manager@example.com",
        )
        employee = _employee()
        employee.id = "emp-risk-stage-change"
        employee.employee_no = "E-RISK-STAGE"
        employee.email = "risk.stage.employee@example.com"
        employee.manager_id = manager.id
        claim = _claim()
        claim.id = "claim-risk-stage-change"
        claim.claim_no = "BX-RISK-STAGE"
        claim.employee_id = employee.id
        claim.approval_stage = "直属领导审批"
        db.add_all([manager, employee, claim])
        db.flush()
        observation = RiskObservationService(db).upsert_observation(
            {
                **_observation_payload("risk:approver:stage-change"),
                "claim_id": claim.id,
                "claim_no": claim.claim_no,
                "subject_key": f"claim:{claim.id}",
            }
        )
        db.commit()
        current_user = CurrentUserContext(
            tenant_id="default",
            username=manager.email,
            name=manager.name,
            role_codes=["manager"],
            is_admin=False,
            employee_no=manager.employee_no,
        )

        claim.approval_stage = "财务审批"
        db.commit()

        with pytest.raises(RiskDispositionPermissionError, match="不再是"):
            RiskDispositionService(db).execute_action(
                observation.id,
                _action("confirm", version=0, request_id="request-stage-changed-001"),
                tenant_id="default",
                actor_id=manager.id,
                actor_name=manager.name,
                current_user=current_user,
            )
        assert db.scalar(select(func.count()).select_from(RiskDispositionEvent)) == 0


def test_waiver_approval_enforces_separation_of_duties_and_replays_snapshot(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setenv("FEW_SHOT_INJECTION_ENABLED", "false")
    with _build_session() as db:
        requester = Employee(
            id="waiver-requester",
            employee_no="W-REQUESTER",
            name="豁免申请人",
            email="waiver.requester@example.com",
            employment_status="在职",
        )
        decider = Employee(
            id="waiver-decider",
            employee_no="W-DECIDER",
            name="豁免决定人",
            email="waiver.decider@example.com",
            employment_status="在职",
        )
        admin_viewer = Employee(
            id="waiver-admin",
            employee_no="W-ADMIN",
            name="仅管理员",
            email="waiver.admin@example.com",
            employment_status="在职",
        )
        inactive_decider = Employee(
            id="waiver-inactive",
            employee_no="W-INACTIVE",
            name="离职财务",
            email="waiver.inactive@example.com",
            employment_status="离职",
        )
        db.add_all([_employee(), _claim(), requester, decider, admin_viewer, inactive_decider])
        db.flush()
        observation = RiskObservationService(db).upsert_observation(
            _observation_payload("risk:waiver:sod")
        )
        observation_id = observation.id
        db.commit()
        service = RiskDispositionService(db)
        requester_user = _user(requester, roles=["finance"])
        decider_user = _user(decider, roles=["finance"])

        service.execute_action(
            observation_id,
            _action("confirm", version=0, request_id="waiver-confirm-request-001"),
            tenant_id="default",
            actor_id=requester.id,
            actor_name=requester.name,
            current_user=requester_user,
        )
        requested = service.execute_action(
            observation_id,
            RiskDispositionActionCreate(
                action="request_waiver",
                expected_version=1,
                request_id="waiver-application-request-001",
                waiver_reason="客户现场临时封控，无法在审批前取得原件。",
                waiver_scope="仅限本次客户拜访的纸质行程确认材料。",
                waiver_expires_at=datetime.now(UTC) + timedelta(days=5),
                waiver_conditions=["三日内补交扫描件", "到期前由财务复核"],
            ),
            tenant_id="default",
            actor_id=requester.id,
            actor_name=requester.name,
            current_user=requester_user,
        )

        assert requested.disposition.lifecycle_status == "waiver_requested"
        assert requested.disposition.waiver_requester_id == requester.id
        assert requested.disposition.waiver_reason == "客户现场临时封控，无法在审批前取得原件。"
        assert requested.disposition.waiver_conditions_json == [
            "三日内补交扫描件",
            "到期前由财务复核",
        ]

        decision = RiskDispositionActionCreate(
            action="approve_waiver",
            expected_version=2,
            request_id="waiver-approve-request-001",
            comment="条件明确且有效期受控，同意本次例外。",
        )
        with pytest.raises(RiskWaiverDecisionPermissionError, match="不能批准或拒绝自己的"):
            service.execute_action(
                observation_id,
                decision,
                tenant_id="default",
                actor_id=requester.id,
                actor_name=requester.name,
                current_user=requester_user,
            )
        with pytest.raises(RiskWaiverDecisionPermissionError, match="明确角色"):
            service.execute_action(
                observation_id,
                decision,
                tenant_id="default",
                actor_id=admin_viewer.id,
                actor_name=admin_viewer.name,
                current_user=_user(admin_viewer, roles=[], is_admin=True),
            )
        with pytest.raises(RiskWaiverDecisionPermissionError, match="不是在职员工"):
            service.execute_action(
                observation_id,
                decision,
                tenant_id="default",
                actor_id=inactive_decider.id,
                actor_name=inactive_decider.name,
                current_user=_user(inactive_decider, roles=["finance"]),
            )
        with pytest.raises(RiskWaiverDecisionPermissionError, match="跨租户"):
            service.execute_action(
                observation_id,
                decision,
                tenant_id="default",
                actor_id=decider.id,
                actor_name=decider.name,
                current_user=_user(decider, roles=["finance"], tenant_id="tenant-b"),
            )

        approved = service.execute_action(
            observation_id,
            decision,
            tenant_id="default",
            actor_id=decider.id,
            actor_name=decider.name,
            current_user=decider_user,
        )
        replayed = service.execute_action(
            observation_id,
            decision,
            tenant_id="default",
            actor_id=decider.id,
            actor_name=decider.name,
            current_user=decider_user,
        )

        assert approved.disposition.lifecycle_status == "waived"
        assert approved.disposition.waiver_decision == "approved"
        assert approved.disposition.waiver_decider_id == decider.id
        assert approved.disposition.waiver_decision_reason == decision.comment
        assert replayed.replayed is True
        assert replayed.event.id == approved.event.id
        assert replayed.disposition.waiver_requester_id == requester.id
        assert replayed.disposition.waiver_decider_id == decider.id
        assert db.scalar(select(func.count()).select_from(RiskDispositionEvent)) == 3


def test_waiver_rejection_is_explicit_and_decisions_require_requested_state(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setenv("FEW_SHOT_INJECTION_ENABLED", "false")
    with _build_session() as db:
        requester = Employee(
            id="waiver-reject-requester",
            employee_no="W-REJECT-REQ",
            name="拒绝流程申请人",
            email="waiver.reject.requester@example.com",
            employment_status="在职",
        )
        decider = Employee(
            id="waiver-reject-decider",
            employee_no="W-REJECT-DEC",
            name="拒绝流程决定人",
            email="waiver.reject.decider@example.com",
            employment_status="在职",
        )
        db.add_all([_employee(), _claim(), requester, decider])
        db.flush()
        observation = RiskObservationService(db).upsert_observation(
            _observation_payload("risk:waiver:rejected")
        )
        observation_id = observation.id
        db.commit()
        service = RiskDispositionService(db)
        requester_user = _user(requester, roles=["finance"])
        decider_user = _user(decider, roles=["executive"])

        with pytest.raises(RiskDispositionConflictError, match="只有待决定"):
            service.execute_action(
                observation_id,
                RiskDispositionActionCreate(
                    action="reject_waiver",
                    expected_version=0,
                    request_id="waiver-reject-without-request",
                    comment="当前没有可拒绝的申请。",
                ),
                tenant_id="default",
                actor_id=decider.id,
                actor_name=decider.name,
                current_user=decider_user,
            )

        service.execute_action(
            observation_id,
            _action("confirm", version=0, request_id="waiver-reject-confirm"),
            tenant_id="default",
            actor_id=requester.id,
            actor_name=requester.name,
            current_user=requester_user,
        )
        service.execute_action(
            observation_id,
            RiskDispositionActionCreate(
                action="request_waiver",
                expected_version=1,
                request_id="waiver-reject-application",
                waiver_reason="希望免除本次高风险拦截。",
                waiver_scope="本次重复票据风险。",
                waiver_expires_at=datetime.now(UTC) + timedelta(days=2),
                waiver_conditions=[],
            ),
            tenant_id="default",
            actor_id=requester.id,
            actor_name=requester.name,
            current_user=requester_user,
        )
        rejected = service.execute_action(
            observation_id,
            RiskDispositionActionCreate(
                action="reject_waiver",
                expected_version=2,
                request_id="waiver-reject-decision",
                comment="缺少足够补偿控制，拒绝豁免。",
            ),
            tenant_id="default",
            actor_id=decider.id,
            actor_name=decider.name,
            current_user=decider_user,
        )

        assert rejected.disposition.lifecycle_status == "waiver_rejected"
        assert rejected.disposition.waiver_decision == "rejected"
        assert rejected.disposition.waiver_decider_id == decider.id
        assert rejected.event.action == "reject_waiver"


def test_waiver_self_decision_endpoint_returns_typed_forbidden_error(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setenv("FEW_SHOT_INJECTION_ENABLED", "false")
    client, session_factory = _build_client()
    requester = Employee(
        id="waiver-api-requester",
        employee_no="W-API-REQ",
        name="接口豁免申请人",
        email="waiver.api.requester@example.com",
        employment_status="在职",
    )
    requester_user = _user(requester, roles=["finance"])
    requester_headers = {
        "X-Auth-Username": requester.email,
        "X-Auth-Name": "Waiver Requester",
        "X-Auth-Employee-No": requester.employee_no,
        "X-Auth-Employee-Id": requester.id,
        "X-Auth-Role-Codes": "finance",
    }
    with session_factory() as db:
        db.add_all([_employee(), _claim(), requester])
        db.flush()
        observation = RiskObservationService(db).upsert_observation(
            _observation_payload("risk:waiver:api-sod")
        )
        observation_id = observation.id
        db.commit()
        service = RiskDispositionService(db)
        service.execute_action(
            observation_id,
            _action("confirm", version=0, request_id="waiver-api-confirm"),
            tenant_id="default",
            actor_id=requester.id,
            actor_name=requester.name,
            current_user=requester_user,
        )
        service.execute_action(
            observation_id,
            RiskDispositionActionCreate(
                action="request_waiver",
                expected_version=1,
                request_id="waiver-api-application",
                waiver_reason="接口申请职责分离验证。",
                waiver_scope="仅限当前风险观察。",
                waiver_expires_at=datetime.now(UTC) + timedelta(days=2),
            ),
            tenant_id="default",
            actor_id=requester.id,
            actor_name=requester.name,
            current_user=requester_user,
        )

    response = client.post(
        f"/api/v1/risk-observations/{observation_id}/disposition/actions",
        headers=requester_headers,
        json={
            "action": "approve_waiver",
            "expected_version": 2,
            "request_id": "waiver-api-self-approve",
            "comment": "尝试自批。",
        },
    )

    assert response.status_code == 403
    assert response.json()["detail"]["code"] == "RISK_WAIVER_DECISION_FORBIDDEN"
    assert "不能批准或拒绝自己的" in response.json()["detail"]["message"]


@pytest.mark.parametrize(
    "action",
    ["false_positive", "request_supplement"],
)
def test_evidence_sensitive_actions_require_server_side_comment(action: str) -> None:
    with pytest.raises(ValueError, match="必须填写 comment"):
        RiskDispositionActionCreate(
            action=action,
            expected_version=0,
            request_id=f"request-comment-{action}",
        )


def test_waiver_request_requires_reason_scope_and_expiry() -> None:
    with pytest.raises(ValueError, match="必须填写原因、范围和有效期"):
        RiskDispositionActionCreate(
            action="request_waiver",
            expected_version=0,
            request_id="request-waiver-fields-missing",
        )


def _action(
    action: str,
    *,
    version: int,
    request_id: str,
) -> RiskDispositionActionCreate:
    waiver_fields: dict[str, object] = {}
    if action == "request_waiver":
        waiver_fields = {
            "waiver_reason": "客户现场条件导致合规材料暂缺",
            "waiver_scope": "仅限本次客户拜访的行程材料",
            "waiver_expires_at": datetime.now(UTC) + timedelta(days=3),
            "waiver_conditions": ["补齐客户确认函", "到期前完成复核"],
        }
    return RiskDispositionActionCreate(
        action=action,
        expected_version=version,
        request_id=request_id,
        comment=("风险处置说明" if action in {"false_positive", "request_supplement"} else None),
        **waiver_fields,
    )


def _build_session() -> Session:
    engine = create_engine(
        "sqlite+pysqlite:///:memory:",
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )
    Base.metadata.create_all(bind=engine)
    factory = sessionmaker(bind=engine, autoflush=False, autocommit=False)
    return factory()


def _build_client() -> tuple[TestClient, sessionmaker[Session]]:
    engine = create_engine(
        "sqlite+pysqlite:///:memory:",
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )
    Base.metadata.create_all(bind=engine)
    factory = sessionmaker(bind=engine, autoflush=False, autocommit=False)
    app = FastAPI()
    app.include_router(risk_observations_router, prefix="/api/v1")
    install_legacy_header_auth_override(app)

    def override_db() -> Generator[Session, None, None]:
        db = factory()
        try:
            yield db
        finally:
            db.close()

    app.dependency_overrides[get_db] = override_db
    return TestClient(app), factory


def _employee() -> Employee:
    return Employee(
        id="emp-risk",
        employee_no="E-RISK",
        name="风险员工",
        email="risk.employee@example.com",
        position="高级专员",
        grade="P6",
    )


def _user(
    employee: Employee,
    *,
    roles: list[str],
    is_admin: bool = False,
    tenant_id: str = "default",
) -> CurrentUserContext:
    return CurrentUserContext(
        username=employee.email,
        name=employee.name,
        role_codes=roles,
        is_admin=is_admin,
        tenant_id=tenant_id,
        employee_no=employee.employee_no,
        employee_id=employee.id,
    )


def _claim() -> ExpenseClaim:
    now = datetime(2026, 7, 16, tzinfo=UTC)
    return ExpenseClaim(
        id="claim-risk-1",
        claim_no="BX-RISK-001",
        employee_id="emp-risk",
        employee_name="风险员工",
        department_id="dept-risk",
        department_name="风控部",
        expense_type="travel",
        reason="客户拜访",
        location="上海",
        amount=Decimal("1200"),
        currency="CNY",
        invoice_count=1,
        occurred_at=now,
        submitted_at=now,
        status="submitted",
        approval_stage="财务审批",
        risk_flags_json=[],
    )


def _observation_payload(observation_key: str) -> dict[str, object]:
    return {
        "observation_key": observation_key,
        "subject_type": "expense_claim",
        "subject_key": "claim:claim-risk-1",
        "subject_label": "BX-RISK-001",
        "claim_id": "claim-risk-1",
        "claim_no": "BX-RISK-001",
        "risk_type": "duplicate_invoice",
        "risk_signal": "duplicate_invoice",
        "title": "重复票据风险",
        "description": "同一票据可能重复报销。",
        "risk_score": 86,
        "risk_level": "high",
        "confidence_score": 0.91,
        "control_stage": "reimbursement",
        "control_mode": "risk_observation",
        "automation_mode": "semi_auto_review",
        "source": "financial_risk_graph",
        "algorithm_version": "financial_risk_graph.v1",
        "contribution_scores": {},
        "baseline": {},
        "evidence": [],
        "graph_node_keys": [],
        "graph_edge_keys": [],
        "policy_refs": [],
        "similar_case_claim_ids": [],
        "ontology_json": {},
        "decision_trace": {},
    }