X-Financial/server/tests/test_agent_asset_release_runtime.py

from __future__ import annotations

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

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

from app.api.deps import get_db
from app.core.agent_enums import AgentAssetDomain, AgentAssetStatus, AgentAssetType
from app.db.base import Base
from app.main import create_app
from app.models.agent_asset import AgentAsset, AgentAssetTestRun, AgentAssetVersion
from app.models.agent_asset_release_telemetry import AgentAssetReleaseObservation
from app.models.audit_log import AuditLog
from app.models.financial_record import ExpenseClaim
from app.models.golden_case import GoldenCase
from app.services.agent_asset_release_guard import (
    AgentAssetReleaseGuardService,
    ReleaseEvaluationInput,
    ReleaseGuardPolicy,
)
from app.services.agent_asset_release_monitor import AgentAssetReleaseMonitor
from app.services.agent_asset_release_monitor_auth import build_release_monitor_signature
from app.services.agent_asset_release_telemetry import (
    AgentAssetReleaseTelemetryService,
    ReleaseObservationInput,
)
from app.services.agent_asset_rule_library import AgentAssetRuleLibraryManager
from app.services.agent_asset_spreadsheet import RISK_RULES_LIBRARY
from app.services.agent_assets import AgentAssetService
from app.services.expense_claim_risk_rule_loader import _is_candidate_route
from app.services.expense_claims import ExpenseClaimService


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


def _manifest(*, name: str, enabled: bool) -> dict:
    return {
        "schema_version": "2.0",
        "rule_code": "risk.release.runtime",
        "name": name,
        "description": name,
        "evaluator": "template_rule",
        "enabled": enabled,
        "applies_to": {
            "domains": ["expense"],
            "business_stages": ["reimbursement"],
            "expense_categories": ["travel"],
        },
        "template_key": "composite_rule_v1",
        "params": {
            "template_key": "composite_rule_v1",
            "field_keys": ["claim.reason"],
            "conditions": [
                {
                    "id": "missing_exception_reason",
                    "operator": "not_contains_any",
                    "fields": ["claim.reason"],
                    "keywords": ["专项审批"],
                }
            ],
            "hit_logic": {"all": ["missing_exception_reason"]},
            "message_template": name,
        },
        "outcomes": {
            "pass": {"action": "continue"},
            "fail": {"severity": "high", "action": "manual_review"},
        },
    }


def _seed_risk_release(
    db: Session,
    manager: AgentAssetRuleLibraryManager,
    *,
    with_golden: bool = True,
    with_report: bool = True,
) -> AgentAsset:
    base_file = "risk.release.runtime.json"
    candidate_file = "risk.release.runtime.v2.json"
    manager.write_rule_library_json(
        library=RISK_RULES_LIBRARY,
        file_name=base_file,
        payload=_manifest(name="基线规则", enabled=True),
    )
    manager.write_rule_library_json(
        library=RISK_RULES_LIBRARY,
        file_name=candidate_file,
        payload=_manifest(name="候选规则", enabled=False),
    )
    asset = AgentAsset(
        id="risk-release-runtime",
        tenant_id="tenant-a",
        scope="tenant",
        asset_type=AgentAssetType.RULE.value,
        code="risk.release.runtime",
        name="基线规则",
        description="基线规则",
        domain=AgentAssetDomain.EXPENSE.value,
        scenario_json=["差旅费"],
        owner="finance",
        status=AgentAssetStatus.ACTIVE.value,
        current_version="v1",
        published_version="v1",
        working_version="v2",
        config_json={
            "tenant_id": "tenant-a",
            "detail_mode": "json_risk",
            "enabled": True,
            "rule_library": RISK_RULES_LIBRARY,
            "rule_document": {"file_name": base_file},
            "revision_draft": {
                "version": "v2",
                "base_version": "v1",
                "generation_status": "completed",
                "rule_document": {"file_name": candidate_file},
            },
        },
    )
    db.add(asset)
    db.add_all(
        [
            AgentAssetVersion(
                asset_id=asset.id,
                tenant_id="tenant-a",
                scope="tenant",
                version=version,
                content=f"# {version}",
                content_type="markdown",
                created_by="pytest",
            )
            for version in ("v1", "v2")
        ]
    )
    if with_report:
        db.add(
            AgentAssetTestRun(
                asset_id=asset.id,
                tenant_id="tenant-a",
                scope="tenant",
                version="v2",
                test_type="report",
                status="passed",
                passed=True,
                summary="candidate report passed",
                created_by="pytest",
            )
        )
    if with_golden:
        db.add(
            GoldenCase(
                case_key="risk-release-runtime-hit",
                rule_code=asset.code,
                name="普通报销应命中",
                values_json={"claim.reason": "普通差旅报销"},
                expected_hit=True,
                expected_severity="high",
                status="active",
            )
        )
    db.commit()
    return asset


def _claim(claim_id: str) -> ExpenseClaim:
    return ExpenseClaim(
        id=claim_id,
        claim_no=f"RE-{claim_id}",
        employee_name="张三",
        department_name="研发部",
        expense_type="travel",
        reason="普通差旅报销",
        location="上海",
        amount=Decimal("1200"),
        currency="CNY",
        invoice_count=0,
        occurred_at=datetime(2026, 7, 16, tzinfo=UTC),
        status="draft",
        approval_stage="待提交",
        risk_flags_json=[],
    )


def _policy() -> ReleaseGuardPolicy:
    return ReleaseGuardPolicy(
        shadow_min_samples=1,
        canary_min_samples=1,
        max_error_rate=0.1,
        min_precision=0.9,
        max_precision_drop=0.05,
        canary_traffic_percent=50,
        recall_gate_enabled=False,
    )


def test_risk_rule_runtime_consumes_shadow_canary_active_and_rollback(
    tmp_path,
    monkeypatch,
) -> None:
    factory = _session_factory()
    manager = AgentAssetRuleLibraryManager(rule_root=tmp_path / "rules")
    with factory() as db:
        asset = _seed_risk_release(db, manager)
        from app.services import expense_claim_platform_risk

        monkeypatch.setattr(
            expense_claim_platform_risk,
            "AgentAssetRuleLibraryManager",
            lambda: manager,
        )
        guard = AgentAssetReleaseGuardService(db, rule_library_manager=manager)
        guard.start_shadow(
            asset.id,
            "v2",
            actor="manager",
            tenant_id="tenant-a",
            policy=_policy(),
        )

        # 启动后改写候选文件，运行时仍使用启动时冻结的可信快照。
        manager.write_rule_library_json(
            library=RISK_RULES_LIBRARY,
            file_name="risk.release.runtime.v2.json",
            payload=_manifest(name="被篡改但不应生效", enabled=True),
        )
        shadow = ExpenseClaimService(db).evaluate_platform_risk_rules(
            _claim("shadow-claim"),
            tenant_id="tenant-a",
            business_stage="reimbursement",
        )
        assert [item["label"] for item in shadow["flags"]] == ["基线规则"]
        assert shadow["shadow_evaluations"] == [
            {
                "asset_id": asset.id,
                "rule_code": asset.code,
                "rule_version": "v2",
                "release_stage": "shadow",
                "hit": True,
                "severity": "high",
            }
        ]
        shadow_sample = db.query(AgentAssetReleaseObservation).filter_by(
            tenant_id="tenant-a",
            asset_id=asset.id,
            stage="shadow",
            version="v2",
        ).one()
        assert shadow_sample.candidate_hit is True
        assert shadow_sample.baseline_hit is True

        guard.record_evaluation(
            asset.id,
            ReleaseEvaluationInput(total=1, failure_count=0, precision=0.99),
            actor="monitor",
            tenant_id="tenant-a",
        )
        guard.promote(asset.id, actor="manager", tenant_id="tenant-a")
        canary_key = next(
            f"canary-{index}"
            for index in range(1000)
            if _is_candidate_route(asset.id, f"canary-{index}", 50)
        )
        canary = ExpenseClaimService(db).evaluate_platform_risk_rules(
            _claim(canary_key),
            tenant_id="tenant-a",
            business_stage="reimbursement",
        )
        assert canary["flags"][0]["label"] == "候选规则"
        assert canary["flags"][0]["rule_version"] == "v2"
        assert canary["flags"][0]["release_stage"] == "canary"
        canary_sample = db.query(AgentAssetReleaseObservation).filter_by(
            tenant_id="tenant-a",
            asset_id=asset.id,
            stage="canary",
            version="v2",
        ).one()
        assert canary_sample.candidate_hit is True

        guard.record_evaluation(
            asset.id,
            ReleaseEvaluationInput(total=1, failure_count=0, precision=0.99),
            actor="monitor",
            tenant_id="tenant-a",
        )
        guard.promote(asset.id, actor="manager", tenant_id="tenant-a")
        active = ExpenseClaimService(db).evaluate_platform_risk_rules(
            _claim("active-claim"),
            tenant_id="tenant-a",
            business_stage="reimbursement",
        )
        assert active["flags"][0]["label"] == "候选规则"
        active_sample = db.query(AgentAssetReleaseObservation).filter_by(
            tenant_id="tenant-a",
            asset_id=asset.id,
            stage="active",
            version="v2",
        ).one()
        assert active_sample.candidate_hit is True
        assert db.get(AgentAsset, asset.id).published_version == "v2"

        result = guard.record_evaluation(
            asset.id,
            ReleaseEvaluationInput(
                total=10,
                failure_count=2,
                precision=0.7,
                baseline_precision=0.99,
            ),
            actor="monitor",
            tenant_id="tenant-a",
        )
        rolled_back = ExpenseClaimService(db).evaluate_platform_risk_rules(
            _claim("rolled-back-claim"),
            tenant_id="tenant-a",
            business_stage="reimbursement",
        )
        refreshed = db.get(AgentAsset, asset.id)
        assert result["release_stage"] == "rolled_back"
        assert rolled_back["flags"][0]["label"] == "基线规则"
        assert refreshed.published_version == "v1"
        assert refreshed.config_json["rule_document"]["file_name"] == ("risk.release.runtime.json")


def test_risk_rule_runtime_blocks_when_candidate_and_stable_snapshots_are_corrupt(
    tmp_path,
    monkeypatch,
) -> None:
    factory = _session_factory()
    manager = AgentAssetRuleLibraryManager(rule_root=tmp_path / "rules")
    with factory() as db:
        asset = _seed_risk_release(db, manager)
        from app.services import expense_claim_platform_risk

        monkeypatch.setattr(
            expense_claim_platform_risk,
            "AgentAssetRuleLibraryManager",
            lambda: manager,
        )
        guard = AgentAssetReleaseGuardService(db, rule_library_manager=manager)
        guard.start_shadow(
            asset.id,
            "v2",
            actor="manager",
            tenant_id="tenant-a",
            policy=_policy(),
        )

        refreshed = db.get(AgentAsset, asset.id)
        config = dict(refreshed.config_json or {})
        state = dict(config["release_guard"])
        state["artifacts"] = {
            version: {**dict(artifact), "sha256": "0" * 64}
            for version, artifact in dict(state["artifacts"]).items()
        }
        config["release_guard"] = state
        refreshed.config_json = config
        db.add(refreshed)
        db.commit()

        review = ExpenseClaimService(db).evaluate_platform_risk_rules(
            _claim("corrupt-release"),
            tenant_id="tenant-a",
            business_stage="reimbursement",
        )

        assert len(review["flags"]) == 1
        assert review["flags"][0]["rule_code"] == asset.code
        assert review["flags"][0]["severity"] == "critical"
        assert review["flags"][0]["action"] == "block"
        assert review["flags"][0]["evidence"]["failed_version"] == "v1"
        assert review["blocking_reasons"]

        result = AgentAssetReleaseMonitor(db).evaluate_current(
            tenant_id="tenant-a",
            asset_id=asset.id,
            actor="artifact-integrity-monitor",
        )
        assert result["status"] == "failed"
        assert result["rolled_back"] is True
        assert result["metrics"]["runtime_failure_count"] == 1


def test_risk_release_fails_closed_without_report_or_golden(tmp_path) -> None:
    manager = AgentAssetRuleLibraryManager(rule_root=tmp_path / "rules")
    with _session_factory()() as db:
        asset = _seed_risk_release(db, manager, with_report=False)
        with pytest.raises(PermissionError, match="测试报告"):
            AgentAssetReleaseGuardService(db, rule_library_manager=manager).start_shadow(
                asset.id,
                "v2",
                actor="manager",
                tenant_id="tenant-a",
            )

    manager = AgentAssetRuleLibraryManager(rule_root=tmp_path / "rules-2")
    with _session_factory()() as db:
        asset = _seed_risk_release(db, manager, with_golden=False)
        with pytest.raises(PermissionError, match="golden case"):
            AgentAssetReleaseGuardService(db, rule_library_manager=manager).start_shadow(
                asset.id,
                "v2",
                actor="manager",
                tenant_id="tenant-a",
            )
        failed = db.query(AgentAssetTestRun).filter_by(test_type="golden").one()
        assert failed.passed is False


def _build_http_client() -> tuple[TestClient, sessionmaker[Session]]:
    factory = _session_factory()
    app = create_app()
    install_legacy_header_auth_override(app)

    def override_db() -> Generator[Session, None, None]:
        with factory() as db:
            yield db

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


def _seed_task_release(factory: sessionmaker[Session]) -> None:
    with factory() as db:
        asset = AgentAsset(
            id="task-release-api",
            tenant_id="tenant-a",
            scope="tenant",
            asset_type=AgentAssetType.TASK.value,
            code="task.release.api",
            name="发布 API 测试",
            domain=AgentAssetDomain.EXPENSE.value,
            owner="manager",
            status=AgentAssetStatus.ACTIVE.value,
            current_version="v1",
            published_version="v1",
            working_version="v2",
            config_json={"tenant_id": "tenant-a"},
        )
        db.add(asset)
        db.add_all(
            [
                AgentAssetVersion(
                    asset_id=asset.id,
                    tenant_id="tenant-a",
                    scope="tenant",
                    version=version,
                    content="{}",
                    content_type="json",
                    created_by="pytest",
                )
                for version in ("v1", "v2")
            ]
        )
        db.commit()


def _seed_global_task_release(factory: sessionmaker[Session]) -> None:
    with factory() as db:
        asset = AgentAsset(
            id="global-task-release-api",
            tenant_id="platform",
            scope="platform",
            asset_type=AgentAssetType.TASK.value,
            code="task.release.global",
            name="平台共享发布 API 测试",
            domain=AgentAssetDomain.EXPENSE.value,
            owner="platform",
            status=AgentAssetStatus.ACTIVE.value,
            current_version="v1",
            published_version="v1",
            working_version="v2",
            config_json={},
        )
        db.add(asset)
        db.add_all(
            [
                AgentAssetVersion(
                    asset_id=asset.id,
                    tenant_id="platform",
                    scope="platform",
                    version=version,
                    content="{}",
                    content_type="json",
                    created_by="pytest",
                )
                for version in ("v1", "v2")
            ]
        )
        db.commit()


def _seed_monitor_risk_release(
    factory: sessionmaker[Session],
    *,
    asset_id: str = "risk-monitor-api",
    tenant_id: str = "tenant-a",
    min_precision: float = 0.9,
) -> None:
    with factory() as db:
        previous_config = {
            "tenant_id": tenant_id,
            "detail_mode": "json_risk",
            "enabled": True,
        }
        db.add(
            AgentAsset(
                id=asset_id,
                tenant_id=tenant_id,
                scope="tenant",
                asset_type=AgentAssetType.RULE.value,
                code=f"risk.{asset_id}",
                name="真实发布监控 API 测试",
                domain=AgentAssetDomain.EXPENSE.value,
                owner="manager",
                status=AgentAssetStatus.ACTIVE.value,
                current_version="v1",
                published_version="v1",
                working_version="v2",
                config_json={
                    **previous_config,
                    "release_guard": {
                        "release_id": f"release-{asset_id}",
                        "stage": "shadow",
                        "candidate_version": "v2",
                        "previous_version": "v1",
                        "started_by": "username:manager",
                        "previous_config": previous_config,
                        "policy": {
                            "shadow_min_samples": 1,
                            "canary_min_samples": 1,
                            "max_error_rate": 1.0,
                            "min_precision": min_precision,
                            "max_precision_drop": 1.0,
                            "canary_traffic_percent": 10,
                        },
                        "history": [],
                    },
                },
            )
        )
        db.commit()


def _append_release_review_samples(
    factory: sessionmaker[Session],
    *,
    asset_id: str,
    labels: list[str],
    tenant_id: str = "tenant-a",
) -> dict:
    with factory() as db:
        asset = db.get(AgentAsset, asset_id)
        assert asset is not None
        state = dict((asset.config_json or {})["release_guard"])
        telemetry = AgentAssetReleaseTelemetryService(db)
        for index, label in enumerate(labels):
            observation = telemetry.record_observation(
                ReleaseObservationInput(
                    tenant_id=tenant_id,
                    asset_id=asset.id,
                    release_id=str(state["release_id"]),
                    stage=str(state["stage"]),  # type: ignore[arg-type]
                    version=str(state["candidate_version"]),
                    rule_code=asset.code,
                    source_key=f"{state['stage']}-sample-{index}",
                    candidate_hit=True,
                    baseline_hit=True,
                )
            )
            telemetry.record_review_label(
                tenant_id=tenant_id,
                observation_id=observation.id,
                label=label,  # type: ignore[arg-type]
                request_id=f"{state['release_id']}:{state['stage']}:review:{index}",
                actor_id="trusted-release-reviewer",
            )
        db.commit()
        return state


def _headers(*, role: str = "manager", tenant: str = "tenant-a") -> dict[str, str]:
    return {
        "x-auth-username": role,
        "x-auth-name": role,
        "x-auth-role-codes": role,
        "x-auth-tenant-id": tenant,
    }


_TEST_MONITOR_SECRET = "release-monitor-test-secret-32-bytes-minimum"


def _monitor_headers(
    *,
    payload: dict,
    state: dict,
    role: str = "manager",
    tenant: str = "tenant-a",
    asset_id: str = "task-release-api",
) -> dict[str, str]:
    timestamp = str(int(time.time()))
    return {
        **_headers(role=role, tenant=tenant),
        "x-release-monitor-timestamp": timestamp,
        "x-release-monitor-signature": build_release_monitor_signature(
            timestamp=timestamp,
            tenant_id=tenant,
            asset_id=asset_id,
            release_id=str(state.get("release_id") or ""),
            stage=str(state.get("stage") or ""),
            payload=payload,
            secret=_TEST_MONITOR_SECRET,
        ),
    }


def test_release_management_http_permissions_tenant_and_auto_rollback(monkeypatch) -> None:
    monkeypatch.setenv("AGENT_RELEASE_MONITOR_SECRET", _TEST_MONITOR_SECRET)
    client, factory = _build_http_client()
    _seed_task_release(factory)
    path = "/api/v1/agent-assets/task-release-api/release"

    assert client.get(path, headers=_headers(role="finance")).status_code == 403
    assert client.get(path, headers=_headers(tenant="tenant-b")).status_code == 404

    started = client.post(
        f"{path}/shadow",
        headers=_headers(),
        json={
            "candidate_version": "v2",
            "policy": {
                "shadow_min_samples": 1,
                "canary_min_samples": 1,
                "max_error_rate": 0.1,
                "min_precision": 0.9,
                "max_precision_drop": 0.05,
                "canary_traffic_percent": 10,
            },
        },
    )
    assert started.status_code == 200
    assert started.json()["stage"] == "shadow"
    release_state = started.json()

    unsigned = client.post(
        f"{path}/evaluations",
        headers=_headers(),
        json={},
    )
    assert unsigned.status_code == 401
    fabricated = {"total": 1, "failure_count": 0, "precision": 1.0}
    assert (
        client.post(
            f"{path}/evaluations",
            headers=_monitor_headers(payload=fabricated, state=release_state),
            json=fabricated,
        ).status_code
        == 422
    )

    _seed_monitor_risk_release(factory)
    risk_path = "/api/v1/agent-assets/risk-monitor-api/release"
    risk_state = _append_release_review_samples(
        factory,
        asset_id="risk-monitor-api",
        labels=["confirmed", "false_positive"],
    )
    trigger: dict = {}
    failed = client.post(
        f"{risk_path}/evaluations",
        headers=_monitor_headers(
            payload=trigger,
            state=risk_state,
            asset_id="risk-monitor-api",
        ),
        json=trigger,
    )
    assert failed.status_code == 200
    assert failed.json()["status"] == "failed"
    assert failed.json()["release_stage"] == "rolled_back"
    assert failed.json()["metrics"]["precision"] == 0.5
    plan = client.get(f"{risk_path}/serving-plan", headers=_headers())
    assert plan.json()["primary_version"] == "v1"

    with factory() as db:
        assert db.query(AuditLog).filter_by(resource_id="risk-monitor-api").count() >= 1


def test_release_review_queue_is_tenant_safe_and_requires_independent_reviewer() -> None:
    client, factory = _build_http_client()
    _seed_monitor_risk_release(factory, asset_id="risk-review-api", min_precision=0.5)
    with factory() as db:
        asset = db.get(AgentAsset, "risk-review-api")
        assert asset is not None
        state = dict((asset.config_json or {})["release_guard"])
        observation = AgentAssetReleaseTelemetryService(db).record_observation(
            ReleaseObservationInput(
                tenant_id="tenant-a",
                asset_id=asset.id,
                release_id=str(state["release_id"]),
                stage="shadow",
                version="v2",
                rule_code=asset.code,
                source_key="claim-review-api",
                candidate_hit=True,
                baseline_hit=True,
            )
        )
        db.commit()
        observation_id = observation.id

    path = "/api/v1/agent-assets/risk-review-api/release/review-queue"
    assert client.get(path, headers=_headers(tenant="tenant-b")).status_code == 404
    queued = client.get(path, headers=_headers())
    assert queued.status_code == 200
    assert queued.json()["pending_total"] == 1
    assert "source_fingerprint" not in queued.text
    item = queued.json()["items"][0]
    assert item["source_document_id"] == "claim-review-api"
    assert item["prediction_blinded"] is True
    assert "candidate_hit" not in item
    assert "baseline_hit" not in item

    self_review = client.post(
        f"{path}/{observation_id}/labels",
        headers={**_headers(), "x-request-id": "self-review"},
        json={"label": "confirmed"},
    )
    assert self_review.status_code == 400
    reviewed = client.post(
        f"{path}/{observation_id}/labels",
        headers={**_headers(role="admin"), "x-request-id": "independent-review"},
        json={"label": "confirmed"},
    )
    assert reviewed.status_code == 200
    assert reviewed.json()["monitor"]["status"] == "passed"
    assert client.get(path, headers=_headers(role="admin")).json()["pending_total"] == 0


def test_only_platform_admin_can_manage_global_release_asset() -> None:
    client, factory = _build_http_client()
    _seed_global_task_release(factory)
    path = "/api/v1/agent-assets/global-task-release-api/release"

    assert client.get(path, headers=_headers(role="manager")).status_code == 404
    assert client.get(path, headers=_headers(role="admin")).status_code == 200
    started = client.post(
        f"{path}/shadow",
        headers=_headers(role="admin"),
        json={"candidate_version": "v2"},
    )
    assert started.status_code == 200
    assert started.json()["stage"] == "shadow"


def test_existing_publish_http_cannot_bypass_shadow_and_canary(
    tmp_path,
    monkeypatch,
) -> None:
    monkeypatch.setenv("AGENT_RELEASE_MONITOR_SECRET", _TEST_MONITOR_SECRET)
    client, factory = _build_http_client()
    manager = AgentAssetRuleLibraryManager(rule_root=tmp_path / "rules")
    with factory() as db:
        _seed_risk_release(db, manager)

    original_init = AgentAssetService.__init__

    def patched_init(self, db, *, current_user=None) -> None:
        original_init(self, db, current_user=current_user)
        self.rule_library_manager = manager

    monkeypatch.setattr(AgentAssetService, "__init__", patched_init)
    asset_path = "/api/v1/agent-assets/risk-release-runtime"
    assert (
        client.post(
            f"{asset_path}/publish",
            headers=_headers(tenant="tenant-b"),
        ).status_code
        == 404
    )
    published = client.post(f"{asset_path}/publish", headers=_headers())

    assert published.status_code == 200
    assert published.json()["published_version"] == "v1"
    assert published.json()["current_version"] == "v1"
    assert published.json()["config_json"]["release_guard"]["stage"] == "shadow"
    assert client.post(f"{asset_path}/activate", headers=_headers()).status_code == 400
    assert (
        client.post(
            f"{asset_path}/risk-rule-enabled",
            headers=_headers(),
            json={"enabled": True},
        ).status_code
        == 400
    )
    assert (
        client.patch(
            asset_path,
            headers=_headers(),
            json={"published_version": "v2"},
        ).status_code
        == 400
    )

    release_path = f"{asset_path}/release"
    for total, expected_stage in ((20, "canary"), (100, "active")):
        release_state = client.get(release_path, headers=_headers()).json()
        _append_release_review_samples(
            factory,
            asset_id="risk-release-runtime",
            labels=["confirmed"] * total,
        )
        payload: dict = {}
        evaluated = client.post(
            f"{release_path}/evaluations",
            headers=_monitor_headers(
                payload=payload,
                state=release_state,
                asset_id="risk-release-runtime",
            ),
            json=payload,
        )
        assert evaluated.status_code == 200
        promoted = client.post(f"{release_path}/promote", headers=_headers())
        assert promoted.status_code == 200
        assert promoted.json()["stage"] == expected_stage

    with factory() as db:
        active = db.get(AgentAsset, "risk-release-runtime")
        assert active is not None
        assert active.published_version == "v2"