from __future__ import annotations

import uuid
from datetime import UTC, datetime, timedelta

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine, delete, select, update
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session

from modelforge_api.domain.lifecycle_contracts import (
    ApprovalDecision,
    ApprovalPolicyCreate,
    ApprovalRequestCreate,
    ApprovalRequestResponse,
    CanaryObservation,
    CleanupExecutionCreate,
    CleanupPlanCreate,
    DeploymentLifecycleState,
    LifecycleEnvironment,
    LifecycleEvidenceBundle,
    LifecycleSubjectCreate,
    PlanExecutionCreate,
    PromotionPlanCreate,
    assert_deployment_transition,
)
from modelforge_api.main import app
from modelforge_api.persistence.models import (
    ArtifactLocation,
    ArtifactLocationRemovalRecord,
    ArtifactSet,
    ArtifactSetMember,
    Base,
    CapabilityDeployment,
    CapabilityResourceEnvelope,
    ComputeNode,
    DeploymentCandidate,
    LifecycleApprovalRequest,
    LifecycleEvent,
    LifecyclePolicyRevision,
    LifecyclePromotionPlan,
    LifecycleRetentionRecord,
    LifecycleRollbackSnapshot,
    LifecycleSubject,
    ModelArtifact,
    RuntimeProbe,
    RuntimeProfile,
    StorageRoot,
)
from modelforge_api.services.lifecycle import LifecycleError, LifecycleService
from modelforge_api.settings import Settings, get_settings
from tests.test_runtime import setup_runtime
from tests.test_serving import approve_and_promote, setup_m5


@pytest.fixture
def session() -> Session:
    engine = create_engine("sqlite+pysqlite:///:memory:")
    Base.metadata.create_all(engine)
    with Session(engine) as value:
        yield value


def lab_evidence(session: Session) -> tuple[LifecycleEvidenceBundle, uuid.UUID]:
    serving, candidate_value, _node = setup_m5(session)
    _approval, deployment = approve_and_promote(serving, candidate_value.id)
    candidate = session.get(DeploymentCandidate, candidate_value.id)
    assert candidate is not None
    artifact_set = session.get(ArtifactSet, candidate.artifact_set_id)
    profile = session.get(RuntimeProfile, candidate.runtime_profile_id)
    probe = session.get(RuntimeProbe, candidate.runtime_probe_id)
    envelope = session.scalar(
        select(CapabilityResourceEnvelope).where(
            CapabilityResourceEnvelope.capability_deployment_id == deployment.id
        )
    )
    assert artifact_set is not None and profile is not None and probe is not None
    assert envelope is not None and profile.runtime_image_digest is not None
    return (
        LifecycleEvidenceBundle(
            artifact_set_id=artifact_set.id,
            model_revision_id=artifact_set.revision_id,
            runtime_profile_id=profile.id,
            runtime_probe_id=probe.id,
            runtime_image_digest=profile.runtime_image_digest,
            capability_deployment_id=deployment.id,
            resource_envelope_id=envelope.id,
            rollback_target_ref="lab-baseline-a",
            integrity="VERIFIED",
            security="APPROVED",
            license="APPROVED",
            runtime="PROVEN",
            evaluation="EVALUATED",
            project_fit="KEEP_LAB",
            production_validation="NOT_REQUIRED",
            scheduler_readiness="READY",
        ),
        deployment.id,
    )


def approved_lab_plan(
    session: Session, *, target: str = "m12-safe-lab"
) -> tuple[LifecycleService, object, object]:
    evidence, deployment_id = lab_evidence(session)
    service = LifecycleService(session)
    subject = service.create_subject(
        LifecycleSubjectCreate(
            target_type="LAB_REHEARSAL",
            target_ref=target,
            environment=LifecycleEnvironment.LAB,
            state=DeploymentLifecycleState.LAB_READY,
        )
    )
    approval = service.request_approval(
        ApprovalRequestCreate(
            policy_key="lab-promotion",
            subject_id=subject.id,
            target_type="LAB_REHEARSAL",
            target_ref=target,
            environment=LifecycleEnvironment.LAB,
            requested_transition=DeploymentLifecycleState.LAB_STABLE,
            evidence=evidence,
            requested_by="requester-a",
            reason="Safe non-production lifecycle rehearsal with exact evidence",
        )
    )
    assert approval.status == "PENDING"
    approval = service.decide_approval(
        approval.id,
        ApprovalDecision(actor="approver-a", reason="Reviewed exact immutable LAB evidence"),
        "approve",
    )
    plan = service.create_plan(
        PromotionPlanCreate(
            approval_request_id=approval.id,
            subject_id=subject.id,
            desired_state=DeploymentLifecycleState.LAB_STABLE,
            migration_class="behavioral",
            candidate_deployment_id=deployment_id,
            rollback_target_ref="lab-baseline-a",
            project_consumers=[],
            affected_identities={
                "runtime_deployments": [str(deployment_id)],
                "resource_envelope_id": str(evidence.resource_envelope_id),
            },
            canary_strategy={
                "mode": "REQUEST",
                "traffic_percent": 0,
                "target_request_count": 2,
                "max_error_rate": 0.0,
                "max_latency_p95_ms": 1000.0,
            },
            health_gates={"worker": "healthy", "scheduler": "ready"},
            automatic_abort_conditions=[
                "HEALTH_FAILURE",
                "ERROR_THRESHOLD",
                "LATENCY_THRESHOLD",
                "CAPACITY_FAILURE",
            ],
            created_by="requester-a",
        )
    )
    plan = service.approve_plan(
        plan.id,
        ApprovalDecision(actor="approver-a", reason="Plan and rollback snapshot inputs reviewed"),
    )
    return service, subject, plan


def test_independent_lifecycle_states_reject_shortcuts() -> None:
    assert_deployment_transition(
        DeploymentLifecycleState.LAB_READY, DeploymentLifecycleState.CANARY
    )
    with pytest.raises(ValueError, match="not allowed"):
        assert_deployment_transition(
            DeploymentLifecycleState.CANDIDATE, DeploymentLifecycleState.STABLE
        )
    with pytest.raises(ValueError, match="not allowed"):
        assert_deployment_transition(
            DeploymentLifecycleState.DEPRECATED, DeploymentLifecycleState.STABLE
        )


def test_policy_revisions_are_immutable_and_do_not_reinterpret_approvals(
    session: Session,
) -> None:
    service = LifecycleService(session)
    service.ensure_defaults()
    original = next(item for item in service.policies() if item.key == "lab-promotion")
    revised = service.create_policy(
        ApprovalPolicyCreate(
            key="lab-promotion",
            scope="LAB_PROMOTION",
            requirements={"integrity": "VERIFIED", "runtime": "PROVEN", "review": "stricter"},
            created_by="policy-owner",
        )
    )
    assert revised.revision == original.revision + 1
    assert revised.id != original.id
    stored = session.get(LifecyclePolicyRevision, original.id)
    assert stored is not None and stored.active is False


def test_deferred_external_validation_blocks_project_production(session: Session) -> None:
    service = LifecycleService(session)
    request = service.request_approval(
        ApprovalRequestCreate(
            policy_key="project-production",
            target_type="CAPABILITY",
            target_ref="vision.embedding@1",
            environment=LifecycleEnvironment.PRODUCTION,
            requested_transition=DeploymentLifecycleState.STABLE,
            evidence=LifecycleEvidenceBundle(
                project_fit="REQUIRES_MORE_EVIDENCE",
                production_validation="DEFERRED_EXTERNAL_VALIDATION",
            ),
            requested_by="operator",
            reason="ExampleVision Vision production gate rehearsal",
        )
    )
    assert request.status == "BLOCKED"
    assert "REQUIRED_EXTERNAL_VALIDATION_NOT_SATISFIED" in request.blockers
    assert "PROJECT_PRODUCTION_FIT_NOT_ELIGIBLE" in request.blockers
    with pytest.raises(LifecycleError, match="cannot be approved"):
        service.decide_approval(
            request.id,
            ApprovalDecision(actor="operator", reason="Cannot override deferred validation"),
            "approve",
        )


def test_approval_becomes_stale_when_referenced_probe_changes(session: Session) -> None:
    evidence, _deployment_id = lab_evidence(session)
    service = LifecycleService(session)
    subject = service.create_subject(
        LifecycleSubjectCreate(
            target_type="LAB_REHEARSAL",
            target_ref="stale-lab",
            environment=LifecycleEnvironment.LAB,
            state=DeploymentLifecycleState.LAB_READY,
        )
    )
    approval = service.request_approval(
        ApprovalRequestCreate(
            policy_key="lab-promotion",
            subject_id=subject.id,
            target_type="LAB_REHEARSAL",
            target_ref="stale-lab",
            environment=LifecycleEnvironment.LAB,
            requested_transition=DeploymentLifecycleState.LAB_STABLE,
            evidence=evidence,
            requested_by="requester",
            reason="Approval staleness must bind the exact runtime probe",
        )
    )
    approval = service.decide_approval(
        approval.id,
        ApprovalDecision(actor="approver", reason="Initial evidence is current and reviewed"),
        "approve",
    )
    probe = session.get(RuntimeProbe, evidence.runtime_probe_id)
    assert probe is not None
    probe.status = "failed"
    session.commit()
    assert service.approval(approval.id).status == "STALE"


def test_expired_and_revoked_approvals_are_terminal(session: Session) -> None:
    evidence, _deployment_id = lab_evidence(session)
    service = LifecycleService(session)
    expired = service.request_approval(
        ApprovalRequestCreate(
            policy_key="lab-promotion",
            target_type="LAB_REHEARSAL",
            target_ref="expired-lab",
            environment=LifecycleEnvironment.LAB,
            requested_transition=DeploymentLifecycleState.LAB_STABLE,
            evidence=evidence,
            requested_by="requester",
            reason="Expired approval fixture uses exact evidence",
            expires_at=datetime.now(UTC) - timedelta(seconds=1),
        )
    )
    assert service.approval(expired.id).status == "EXPIRED"

    subject = service.create_subject(
        LifecycleSubjectCreate(
            target_type="LAB_REHEARSAL",
            target_ref="revoke-lab",
            environment=LifecycleEnvironment.LAB,
            state=DeploymentLifecycleState.LAB_READY,
        )
    )
    current = service.request_approval(
        ApprovalRequestCreate(
            policy_key="lab-promotion",
            subject_id=subject.id,
            target_type="LAB_REHEARSAL",
            target_ref="revoke-lab",
            environment=LifecycleEnvironment.LAB,
            requested_transition=DeploymentLifecycleState.LAB_STABLE,
            evidence=evidence,
            requested_by="requester",
            reason="Revocation fixture uses exact evidence",
        )
    )
    current = service.decide_approval(
        current.id,
        ApprovalDecision(actor="approver", reason="Exact LAB evidence reviewed before revocation"),
        "approve",
    )
    revoked = service.decide_approval(
        current.id,
        ApprovalDecision(
            actor="approver", reason="Approval intentionally revoked before execution"
        ),
        "revoke",
    )
    assert revoked.status == "REVOKED"


def test_lab_canary_requires_review_then_commits_and_rolls_back(session: Session) -> None:
    service, subject_value, plan = approved_lab_plan(session)
    operation = service.execute_plan(
        plan.id,
        PlanExecutionCreate(
            executor="executor-a",
            idempotency_key="lab-positive-operation",
            expected_subject_version=subject_value.version,
        ),
    )
    assert operation.stage == "VERIFYING"
    assert operation.canary is not None and operation.canary["status"] == "RUNNING"
    current_subject = next(item for item in service.subjects() if item.id == subject_value.id)
    assert current_subject.state == "CANARY"
    operation = service.observe_canary(
        operation.id,
        CanaryObservation(
            request_count=2,
            error_count=0,
            latency_p95_ms=125.0,
            capability_health="HEALTHY",
            scheduler_ready=True,
        ),
    )
    assert operation.canary is not None
    assert operation.canary["status"] == "READY_FOR_PROMOTION_REVIEW"
    assert (
        next(item for item in service.subjects() if item.id == subject_value.id).state == "CANARY"
    )
    operation = service.commit_operation(
        operation.id,
        ApprovalDecision(actor="executor-a", reason="Canary evidence reviewed; commit LAB state"),
    )
    assert operation.stage == "COMMITTED"
    assert (
        next(item for item in service.subjects() if item.id == subject_value.id).state
        == "LAB_STABLE"
    )
    assert session.scalar(select(LifecycleRetentionRecord)) is not None
    rolled_back = service.rollback_operation(
        operation.id,
        ApprovalDecision(
            actor="executor-a", reason="Intentional non-production rollback rehearsal"
        ),
    )
    assert rolled_back.stage == "ROLLED_BACK"
    assert rolled_back.rollback_duration_ms is not None
    assert (
        next(item for item in service.subjects() if item.id == subject_value.id).state
        == "LAB_READY"
    )


def test_canary_auto_abort_restores_snapshot(session: Session) -> None:
    service, subject_value, plan = approved_lab_plan(session, target="abort-lab")
    operation = service.execute_plan(
        plan.id,
        PlanExecutionCreate(
            executor="executor-a",
            idempotency_key="lab-abort-operation",
            expected_subject_version=subject_value.version,
        ),
    )
    operation = service.observe_canary(
        operation.id,
        CanaryObservation(
            request_count=1,
            error_count=1,
            latency_p95_ms=50.0,
            capability_health="HEALTHY",
            scheduler_ready=True,
        ),
    )
    assert operation.stage == "ROLLED_BACK"
    assert operation.failure_code == "ERROR_THRESHOLD"
    assert (
        next(item for item in service.subjects() if item.id == subject_value.id).state
        == "LAB_READY"
    )


def test_restart_reconciliation_and_cas_are_conservative(session: Session) -> None:
    service, subject_value, plan = approved_lab_plan(session, target="restart-lab")
    with pytest.raises(LifecycleError, match="changed concurrently"):
        service.execute_plan(
            plan.id,
            PlanExecutionCreate(
                executor="executor-a",
                idempotency_key="wrong-version-operation",
                expected_subject_version=subject_value.version + 1,
            ),
        )
    operation = service.execute_plan(
        plan.id,
        PlanExecutionCreate(
            executor="executor-a",
            idempotency_key="restart-operation",
            expected_subject_version=subject_value.version,
            rehearsal_pause_stage="PREPARING",
        ),
    )
    assert operation.stage == "PREPARING"
    assert service.reconcile_incomplete() == 1
    assert service.operations()[0].stage == "ROLLED_BACK"
    assert (
        next(item for item in service.subjects() if item.id == subject_value.id).state
        == "LAB_READY"
    )


def test_approval_transition_and_exact_rollback_target_bind_the_plan(
    session: Session,
) -> None:
    service, subject_value, approved = approved_lab_plan(session, target="bound-plan")
    with pytest.raises(LifecycleError) as transition_error:
        service.create_plan(
            PromotionPlanCreate(
                approval_request_id=approved.approval_request_id,
                subject_id=subject_value.id,
                desired_state=DeploymentLifecycleState.DEPRECATED,
                migration_class="transparent",
                candidate_deployment_id=approved.candidate_deployment_id,
                rollback_target_ref=approved.rollback_target_ref,
                created_by="requester-a",
            )
        )
    assert transition_error.value.code == "APPROVAL_TRANSITION_MISMATCH"
    with pytest.raises(LifecycleError) as rollback_error:
        service.create_plan(
            PromotionPlanCreate(
                approval_request_id=approved.approval_request_id,
                subject_id=subject_value.id,
                desired_state=DeploymentLifecycleState.LAB_STABLE,
                migration_class="behavioral",
                candidate_deployment_id=approved.candidate_deployment_id,
                rollback_target_ref="different-unapproved-target",
                canary_strategy={"mode": "REQUEST", "target_request_count": 1},
                created_by="requester-a",
            )
        )
    assert rollback_error.value.code == "APPROVAL_ROLLBACK_TARGET_MISMATCH"


def test_competing_promotion_plans_are_serialized_by_database_cas(session: Session) -> None:
    service, subject_value, first_plan = approved_lab_plan(session, target="promotion-race")
    first_approval = session.get(LifecycleApprovalRequest, first_plan.approval_request_id)
    assert first_approval is not None
    evidence = LifecycleEvidenceBundle.model_validate(first_approval.evidence_snapshot["declared"])
    second_approval = service.request_approval(
        ApprovalRequestCreate(
            policy_key="lab-promotion",
            subject_id=subject_value.id,
            target_type="LAB_REHEARSAL",
            target_ref=subject_value.target_ref,
            environment=LifecycleEnvironment.LAB,
            requested_transition=DeploymentLifecycleState.LAB_STABLE,
            evidence=evidence,
            requested_by="requester-b",
            reason="Second operator prepares a competing exact LAB transition",
        )
    )
    second_approval = service.decide_approval(
        second_approval.id,
        ApprovalDecision(actor="approver-b", reason="Competing plan evidence independently reviewed"),
        "approve",
    )
    second_plan = service.create_plan(
        PromotionPlanCreate(
            approval_request_id=second_approval.id,
            subject_id=subject_value.id,
            desired_state=DeploymentLifecycleState.LAB_STABLE,
            migration_class="behavioral",
            candidate_deployment_id=first_plan.candidate_deployment_id,
            rollback_target_ref=first_plan.rollback_target_ref,
            affected_identities={"race_contender": "b"},
            canary_strategy={"mode": "REQUEST", "target_request_count": 1},
            created_by="requester-b",
        )
    )
    second_plan = service.approve_plan(
        second_plan.id,
        ApprovalDecision(actor="approver-b", reason="Second immutable plan approved for CAS test"),
    )
    first = service.execute_plan(
        first_plan.id,
        PlanExecutionCreate(
            executor="executor-a",
            idempotency_key="promotion-race-winner",
            expected_subject_version=subject_value.version,
        ),
    )
    retry = service.execute_plan(
        first_plan.id,
        PlanExecutionCreate(
            executor="executor-a",
            idempotency_key="promotion-race-winner",
            expected_subject_version=subject_value.version,
        ),
    )
    assert retry.id == first.id
    assert len([item for item in service.operations() if item.id == first.id]) == 1
    with pytest.raises(LifecycleError) as conflict:
        service.execute_plan(
            second_plan.id,
            PlanExecutionCreate(
                executor="executor-b",
                idempotency_key="promotion-race-loser",
                expected_subject_version=subject_value.version,
            ),
        )
    assert conflict.value.code == "VERSION_CONFLICT"
    stored_second = session.get(LifecyclePromotionPlan, second_plan.id)
    assert stored_second is not None and stored_second.status == "APPROVED"


def test_rollback_race_preserves_newer_subject_state_and_reports_failure(
    session: Session,
) -> None:
    service, subject_value, plan = approved_lab_plan(session, target="rollback-race")
    operation = service.execute_plan(
        plan.id,
        PlanExecutionCreate(
            executor="executor-a",
            idempotency_key="rollback-race-operation",
            expected_subject_version=subject_value.version,
        ),
    )
    session.execute(
        update(LifecycleSubject)
        .where(LifecycleSubject.id == subject_value.id)
        .values(version=LifecycleSubject.version + 1)
    )
    session.commit()
    failed = service.rollback_operation(
        operation.id,
        ApprovalDecision(actor="executor-a", reason="Exercise rollback versus newer transition"),
    )
    current = next(item for item in service.subjects() if item.id == subject_value.id)
    assert failed.stage == "FAILED" and failed.failure_code == "ROLLBACK_VERSION_CONFLICT"
    assert current.state == "CANARY" and current.version == 3


def test_deprecation_supersession_archive_and_retention_expiry_are_independent(
    session: Session,
) -> None:
    base_evidence, deployment_id = lab_evidence(session)
    assert base_evidence.artifact_set_id is not None
    rollback_ref = str(base_evidence.artifact_set_id)
    evidence = base_evidence.model_copy(update={"rollback_target_ref": rollback_ref})
    service = LifecycleService(session)
    subject = service.create_subject(
        LifecycleSubjectCreate(
            target_type="LAB_REHEARSAL",
            target_ref="deprecation-retention-lab",
            environment=LifecycleEnvironment.LAB,
            state=DeploymentLifecycleState.LAB_READY,
        )
    )

    def approved_transition(
        desired: DeploymentLifecycleState, suffix: str
    ) -> ApprovalRequestResponse:
        approval = service.request_approval(
            ApprovalRequestCreate(
                policy_key="lab-promotion",
                subject_id=subject.id,
                target_type="LAB_REHEARSAL",
                target_ref=subject.target_ref,
                environment=LifecycleEnvironment.LAB,
                requested_transition=desired,
                evidence=evidence,
                requested_by=f"requester-{suffix}",
                reason=f"Exact evidence for logical {desired} transition without binary deletion",
            )
        )
        return service.decide_approval(
            approval.id,
            ApprovalDecision(
                actor=f"approver-{suffix}", reason="Lifecycle and retention effects reviewed"
            ),
            "approve",
        )

    deprecation_approval = approved_transition(DeploymentLifecycleState.DEPRECATED, "dep")
    deprecation_plan = service.create_plan(
        PromotionPlanCreate(
            approval_request_id=deprecation_approval.id,
            subject_id=subject.id,
            desired_state=DeploymentLifecycleState.DEPRECATED,
            migration_class="transparent",
            candidate_deployment_id=deployment_id,
            rollback_target_ref=rollback_ref,
            affected_identities={"superseded_by_ref": "replacement-lab-revision"},
            created_by="requester-dep",
        )
    )
    deprecation_plan = service.approve_plan(
        deprecation_plan.id,
        ApprovalDecision(actor="approver-dep", reason="Deprecation plan reviewed and frozen"),
    )
    deprecation_operation = service.execute_plan(
        deprecation_plan.id,
        PlanExecutionCreate(
            executor="executor-dep",
            idempotency_key="deprecation-retention-operation",
            expected_subject_version=subject.version,
        ),
    )
    service.commit_operation(
        deprecation_operation.id,
        ApprovalDecision(actor="executor-dep", reason="Commit logical deprecation only"),
    )
    deprecated = next(item for item in service.subjects() if item.id == subject.id)
    artifact_set = session.get(ArtifactSet, base_evidence.artifact_set_id)
    retention = session.scalar(
        select(LifecycleRetentionRecord).where(
            LifecycleRetentionRecord.target_ref == rollback_ref
        )
    )
    assert deprecated.state == "DEPRECATED"
    assert deprecated.superseded_by_ref == "replacement-lab-revision"
    assert artifact_set is not None and artifact_set.status == "verified"
    assert retention is not None and retention.state == "ROLLBACK_RETAINED"

    protected = service.create_cleanup_plan(
        CleanupPlanCreate(
            target_type="ARTIFACT_SET",
            target_ref=base_evidence.artifact_set_id,
            action="ARCHIVE_METADATA",
            created_by="operator",
        )
    )
    protected_retention = next(
        item for item in protected.dependencies if item["type"] == "RETENTION_POLICY"
    )
    assert protected.retention_state == "ROLLBACK_RETAINED"
    assert protected_retention["blocking"] is True
    retention.retained_until = datetime.now(UTC) - timedelta(seconds=1)
    session.commit()
    expired = service.create_cleanup_plan(
        CleanupPlanCreate(
            target_type="ARTIFACT_SET",
            target_ref=base_evidence.artifact_set_id,
            action="ARCHIVE_METADATA",
            created_by="operator",
        )
    )
    expired_retention = next(
        item for item in expired.dependencies if item["type"] == "RETENTION_POLICY"
    )
    assert expired.retention_state == "ARCHIVABLE"
    assert expired_retention["blocking"] is False
    assert expired.status == "BLOCKED"  # independent active dependencies still win

    archive_approval = approved_transition(DeploymentLifecycleState.ARCHIVED, "archive")
    archive_plan = service.create_plan(
        PromotionPlanCreate(
            approval_request_id=archive_approval.id,
            subject_id=subject.id,
            desired_state=DeploymentLifecycleState.ARCHIVED,
            migration_class="transparent",
            candidate_deployment_id=deployment_id,
            rollback_target_ref=rollback_ref,
            affected_identities={"superseded_by_ref": "replacement-lab-revision"},
            created_by="requester-archive",
        )
    )
    archive_plan = service.approve_plan(
        archive_plan.id,
        ApprovalDecision(actor="approver-archive", reason="Logical archive plan reviewed"),
    )
    archive_operation = service.execute_plan(
        archive_plan.id,
        PlanExecutionCreate(
            executor="executor-archive",
            idempotency_key="logical-archive-operation",
            expected_subject_version=deprecated.version,
        ),
    )
    service.commit_operation(
        archive_operation.id,
        ApprovalDecision(actor="executor-archive", reason="Commit logical archive only"),
    )
    archived = next(item for item in service.subjects() if item.id == subject.id)
    assert archived.state == "ARCHIVED"
    assert artifact_set.status == "verified"
    event_types = {item.event_type for item in service.events(1000)}
    assert {"DEPLOYMENT_DEPRECATED", "DEPLOYMENT_ARCHIVED", "RETENTION_APPLIED"} <= event_types


def test_approved_plan_is_immutable(session: Session) -> None:
    _service, _subject, plan_value = approved_lab_plan(session, target="immutable-plan")
    plan = session.get(LifecyclePromotionPlan, plan_value.id)
    assert plan is not None
    plan.desired_state = "STABLE"
    with pytest.raises(ValueError, match="immutable approved fields"):
        session.commit()
    session.rollback()


def test_cleanup_dry_run_blocks_dependencies_and_rechecks_safe_location(
    session: Session,
) -> None:
    _runtime, artifact_set, _node, artifacts = setup_runtime(session)
    service = LifecycleService(session)
    blocked = service.create_cleanup_plan(
        CleanupPlanCreate(
            target_type="ARTIFACT_SET",
            target_ref=artifact_set.id,
            action="ARCHIVE_METADATA",
            created_by="operator",
        )
    )
    assert blocked.status == "BLOCKED"
    assert blocked.reclaimable_bytes == artifact_set.total_size_bytes

    root = session.scalar(select(StorageRoot))
    assert root is not None
    orphan = ModelArtifact(
        revision_id=artifacts[0].revision_id,
        filename="m12-unused-test.safetensors",
        artifact_type="weights",
        serialization_format="safetensors",
        sha256="d" * 64,
        size_bytes=4096,
        security_status="verified",
        license_status="approved",
        status="verified",
        quarantined=False,
    )
    session.add(orphan)
    session.flush()
    location = ArtifactLocation(
        artifact_id=orphan.id,
        storage_root_id=root.id,
        relative_path="m12/unused-test.safetensors",
        status="verified",
        size_bytes=4096,
        observed_sha256=orphan.sha256,
    )
    session.add(location)
    session.commit()
    ready = service.create_cleanup_plan(
        CleanupPlanCreate(
            target_type="ARTIFACT_LOCATION",
            target_ref=location.id,
            action="RECORD_LOCATION_REMOVAL",
            created_by="operator",
        )
    )
    assert ready.status == "READY" and ready.reclaimable_bytes == 4096
    executed = service.execute_cleanup(
        ready.id,
        CleanupExecutionCreate(executor="operator", confirm=True, physical_removal_confirmed=True),
    )
    assert executed.status == "EXECUTED"
    assert location.status == "removed" and orphan.status == "missing"
    assert session.scalar(select(ArtifactLocationRemovalRecord)) is not None


def test_lifecycle_events_are_append_only(session: Session) -> None:
    service = LifecycleService(session)
    service.ensure_defaults()
    event = session.scalar(select(LifecycleEvent))
    assert event is not None
    event.reason = "tampered"
    with pytest.raises(ValueError, match="append-only"):
        session.commit()
    session.rollback()


def test_lifecycle_routes_deny_capability_credentials() -> None:
    app.dependency_overrides[get_settings] = lambda: Settings(operator_api_key="operator-secret")
    try:
        client = TestClient(app)
        assert client.get("/api/v1/admin/lifecycle/approval-policies").status_code == 401
        response = client.get(
            "/api/v1/admin/lifecycle/approval-policies",
            headers={"Authorization": "Bearer project-capability-secret"},
        )
        assert response.status_code == 401
    finally:
        app.dependency_overrides.clear()


def test_blocked_approval_is_audited_without_payload_content(session: Session) -> None:
    service = LifecycleService(session)
    result = service.request_approval(
        ApprovalRequestCreate(
            policy_key="capability-production",
            target_type="CAPABILITY",
            target_ref="document.ocr@1",
            environment=LifecycleEnvironment.PRODUCTION,
            requested_transition=DeploymentLifecycleState.STABLE,
            evidence=LifecycleEvidenceBundle(),
            requested_by="operator",
            reason="OCR production approval blocker rehearsal",
        )
    )
    event = session.scalar(select(LifecycleEvent).where(LifecycleEvent.change_id == str(result.id)))
    approval = session.get(LifecycleApprovalRequest, result.id)
    assert event is not None and approval is not None
    assert event.to_state == "BLOCKED"
    assert "payload" not in event.details and "content" not in event.details


def test_declared_green_states_cannot_replace_exact_production_evidence(session: Session) -> None:
    evidence, _deployment_id = lab_evidence(session)
    forged = evidence.model_copy(
        update={
            "model_revision_id": uuid.uuid4(),
            "runtime_image_digest": "sha256:" + "f" * 64,
            "security": "APPROVED",
            "license": "APPROVED",
            "evaluation": "PROMOTION_ELIGIBLE",
            "scheduler_readiness": "READY",
            "critical_regressions": 0,
        }
    )
    result = LifecycleService(session).request_approval(
        ApprovalRequestCreate(
            policy_key="capability-production",
            target_type="CAPABILITY",
            target_ref="vision.embedding@1",
            environment=LifecycleEnvironment.PRODUCTION,
            requested_transition=DeploymentLifecycleState.STABLE,
            evidence=forged,
            requested_by="operator",
            reason="Attempt to prove that claimed green labels cannot replace database evidence",
        )
    )
    assert result.status == "BLOCKED"
    assert "MODEL_REVISION_ARTIFACT_SET_MISMATCH" in result.blockers
    assert "RUNTIME_IMAGE_DIGEST_MISMATCH" in result.blockers
    assert "MISSING_PRODUCTION_EXECUTION_APPROVAL" in result.blockers
    assert "MISSING_EVALUATION_RUNS" in result.blockers
    assert "MISSING_SCHEDULER_EVIDENCE_REVISION" in result.blockers


def test_database_prevents_two_production_stable_deployments(session: Session) -> None:
    _evidence, deployment_id = lab_evidence(session)
    current = session.get(CapabilityDeployment, deployment_id)
    assert current is not None and current.production and current.status == "stable"
    duplicate = CapabilityDeployment(
        capability_contract_id=current.capability_contract_id,
        deployment_candidate_id=current.deployment_candidate_id,
        production_approval_id=current.production_approval_id,
        execution_approval_id=current.execution_approval_id,
        embedding_space_id=current.embedding_space_id,
        artifact_set_id=current.artifact_set_id,
        runtime_profile_id=current.runtime_profile_id,
        compute_node_id=current.compute_node_id,
        accelerator_id=current.accelerator_id,
        channel="stable",
        status="stable",
        production=True,
        health_status="healthy",
        routing_weight=100,
        fallback_policy={},
        residency_policy="on_demand",
        keep_warm_seconds=0,
        max_concurrency=1,
        max_queue_depth=1,
        config_fingerprint="f" * 64,
        provenance={},
        rollback_policy={},
    )
    session.add(duplicate)
    with pytest.raises(IntegrityError):
        session.commit()
    session.rollback()


def test_missing_rollback_snapshot_fails_conservatively(session: Session) -> None:
    service, subject_value, plan = approved_lab_plan(session, target="rollback-failure-lab")
    operation = service.execute_plan(
        plan.id,
        PlanExecutionCreate(
            executor="executor-a",
            idempotency_key="rollback-failure-operation",
            expected_subject_version=subject_value.version,
        ),
    )
    session.execute(
        delete(LifecycleRollbackSnapshot).where(
            LifecycleRollbackSnapshot.promotion_plan_id == plan.id
        )
    )
    session.commit()
    failed = service.rollback_operation(
        operation.id,
        ApprovalDecision(actor="executor-a", reason="Exercise unavailable rollback target path"),
    )
    subject = next(item for item in service.subjects() if item.id == subject_value.id)
    assert failed.stage == "FAILED" and failed.failure_code == "ROLLBACK_FAILED"
    assert subject.state == "MANUAL_INTERVENTION_REQUIRED"


def test_cleanup_plan_becomes_stale_when_dependency_appears(session: Session) -> None:
    _runtime, artifact_set, _node, artifacts = setup_runtime(session)
    root = session.scalar(select(StorageRoot))
    assert root is not None
    orphan = ModelArtifact(
        revision_id=artifacts[0].revision_id,
        filename="m12-cleanup-race.safetensors",
        artifact_type="weights",
        serialization_format="safetensors",
        sha256="e" * 64,
        size_bytes=2048,
        security_status="verified",
        license_status="approved",
        status="verified",
        quarantined=False,
    )
    session.add(orphan)
    session.flush()
    location = ArtifactLocation(
        artifact_id=orphan.id,
        storage_root_id=root.id,
        relative_path="m12/cleanup-race.safetensors",
        status="verified",
        size_bytes=2048,
        observed_sha256=orphan.sha256,
    )
    session.add(location)
    session.commit()
    service = LifecycleService(session)
    plan = service.create_cleanup_plan(
        CleanupPlanCreate(
            target_type="ARTIFACT_LOCATION",
            target_ref=location.id,
            action="RECORD_LOCATION_REMOVAL",
            created_by="operator",
        )
    )
    assert plan.status == "READY"
    session.add(
        ArtifactSetMember(
            artifact_set_id=artifact_set.id,
            artifact_id=orphan.id,
            ordinal=artifact_set.file_count,
            required=False,
        )
    )
    session.commit()
    with pytest.raises(LifecycleError) as error:
        service.execute_cleanup(
            plan.id,
            CleanupExecutionCreate(
                executor="operator", confirm=True, physical_removal_confirmed=True
            ),
        )
    assert error.value.code == "CLEANUP_PLAN_STALE"
    stored = next(item for item in service.cleanup_plans() if item.id == plan.id)
    assert stored.status == "STALE" and location.status == "verified"


def test_node_specific_location_removal_does_not_mark_global_artifact_missing(
    session: Session,
) -> None:
    _runtime, _artifact_set, _node, artifacts = setup_runtime(session)
    first_root = session.scalar(select(StorageRoot))
    assert first_root is not None
    second_node = ComputeNode(key="m12-replica", hostname="m12-replica", liveness_state="online")
    session.add(second_node)
    session.flush()
    second_root = StorageRoot(
        compute_node_id=second_node.id,
        name="replica-models",
        path="/replica/models",
        agent_path="/data/artifacts/model-registry",
        status="ready",
        writable=True,
    )
    orphan = ModelArtifact(
        revision_id=artifacts[0].revision_id,
        filename="m12-multi-node.safetensors",
        artifact_type="weights",
        serialization_format="safetensors",
        sha256="9" * 64,
        size_bytes=1024,
        security_status="verified",
        license_status="approved",
        status="verified",
        quarantined=False,
    )
    session.add_all([second_root, orphan])
    session.flush()
    first_location = ArtifactLocation(
        artifact_id=orphan.id,
        storage_root_id=first_root.id,
        relative_path="m12/multi-node.safetensors",
        status="verified",
        size_bytes=orphan.size_bytes,
        observed_sha256=orphan.sha256,
    )
    second_location = ArtifactLocation(
        artifact_id=orphan.id,
        storage_root_id=second_root.id,
        relative_path="m12/multi-node.safetensors",
        status="verified",
        size_bytes=orphan.size_bytes,
        observed_sha256=orphan.sha256,
    )
    session.add_all([first_location, second_location])
    session.commit()
    service = LifecycleService(session)
    plan = service.create_cleanup_plan(
        CleanupPlanCreate(
            target_type="ARTIFACT_LOCATION",
            target_ref=second_location.id,
            action="RECORD_LOCATION_REMOVAL",
            created_by="operator",
        )
    )
    assert plan.status == "READY"
    assert (
        next(item for item in plan.dependencies if item["type"] == "OTHER_ARTIFACT_LOCATIONS")[
            "ref"
        ]
        == "1"
    )
    service.execute_cleanup(
        plan.id,
        CleanupExecutionCreate(executor="operator", confirm=True, physical_removal_confirmed=True),
    )
    assert second_location.status == "removed"
    assert first_location.status == "verified" and orphan.status == "verified"