ModelForge/backend/tests/test_observability_m14.py

from __future__ import annotations

import uuid
from datetime import UTC, datetime, timedelta

import pytest
from sqlalchemy import create_engine, func, select
from sqlalchemy.orm import Session

from modelforge_api.domain.observability import (
    AlertAction,
    AlertState,
    MaintenanceWindowCreate,
    MetricDefinition,
    MetricRegistry,
    SLOPolicyCreate,
    SLOState,
    TelemetryType,
    metrics,
)
from modelforge_api.persistence.models import (
    Accelerator,
    AcceleratorTelemetryLatest,
    AlertHistoryEvent,
    AlertRuleRevision,
    Base,
    CapacityAggregate,
    CapacitySnapshot,
    ComputeNode,
    GatewayRequest,
    OperationalAlert,
    OperationalIncident,
    ServiceLevelIndicator,
    SLOPolicyRevision,
    StorageRoot,
    StorageVolumeState,
)
from modelforge_api.services.observability import ObservabilityService
from modelforge_api.settings import Settings


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


def service(session: Session) -> ObservabilityService:
    return ObservabilityService(
        session,
        Settings(observability_monitor_enabled=False),
        actor="m14-test",
    )


def production_node(session: Session, *, liveness: str = "offline") -> ComputeNode:
    node = ComputeNode(
        key=f"m14-node-{uuid.uuid4()}",
        hostname="gpu_node.example.test",
        display_name="GPU Node",
        identity_source="node_agent",
        status="active",
        inventory={},
        enabled=True,
        production_eligible=True,
        lab_eligible=True,
        benchmark_eligible=True,
        liveness_state=liveness,
        last_heartbeat_at=datetime.now(UTC),
        total_ram_bytes=64 * 1024**3,
    )
    session.add(node)
    session.flush()
    return node


def test_metric_contract_rejects_unbounded_labels_and_exports_histogram() -> None:
    registry = MetricRegistry()
    registry.define(
        MetricDefinition(
            name="modelforge_test_duration_seconds",
            type=TelemetryType.HISTOGRAM,
            help="Bounded test duration",
            labels=("route_class",),
        )
    )
    registry.observe("modelforge_test_duration_seconds", {"route_class": "gateway"}, 0.25)

    output = registry.render()

    assert "# TYPE modelforge_test_duration_seconds histogram" in output
    assert 'modelforge_test_duration_seconds_bucket{le="0.25",route_class="gateway"} 1' in output
    assert 'modelforge_test_duration_seconds_bucket{le="+Inf",route_class="gateway"} 1' in output
    assert 'modelforge_test_duration_seconds_sum{route_class="gateway"} 0.25' in output
    with pytest.raises(ValueError, match="bounded and non-sensitive"):
        registry.define(
            MetricDefinition(
                name="modelforge_invalid_total",
                type=TelemetryType.COUNTER,
                help="Invalid metric",
                labels=("request_id",),
            )
        )
    counter = MetricRegistry()
    counter.define(
        MetricDefinition(
            name="modelforge_test_requests_total",
            type=TelemetryType.COUNTER,
            help="Bounded request outcomes",
            labels=("status",),
        )
    )
    counter.increment("modelforge_test_requests_total", {"status": "ok"}, 2)
    assert counter.samples("modelforge_test_requests_total") == [({"status": "ok"}, 2)]
    with pytest.raises(ValueError, match="requires labels"):
        counter.increment("modelforge_test_requests_total", {"payload": "secret"})


def test_defaults_are_typed_versioned_and_lab_has_no_error_budget(session: Session) -> None:
    subject = service(session)
    subject.ensure_defaults()

    assert session.scalar(select(func.count()).select_from(ServiceLevelIndicator)) == 11
    assert session.scalar(select(func.count()).select_from(SLOPolicyRevision)) == 9
    # 12 M14 operational rules plus the five M15 recovery rules.
    assert session.scalar(select(func.count()).select_from(AlertRuleRevision)) == 17
    recovery_rules = {
        item.key: item.alert_type
        for item in session.scalars(
            select(AlertRuleRevision).where(AlertRuleRevision.key.like("recovery.%"))
        )
    }
    assert recovery_rules == {
        "recovery.backup-stale": "BACKUP_STALE",
        "recovery.backup-failed": "BACKUP_FAILED",
        "recovery.backup-verification-failed": "BACKUP_VERIFICATION_FAILED",
        "recovery.restore-failed": "RESTORE_FAILED",
        "recovery.readiness-degraded": "RECOVERY_READINESS_DEGRADED",
    }

    original = session.scalar(
        select(SLOPolicyRevision).where(SLOPolicyRevision.key == "gateway.production")
    )
    assert original is not None
    revised = subject.create_policy(
        SLOPolicyCreate(
            key=original.key,
            sli_definition_id=original.sli_definition_id,
            objective=0.995,
            rolling_window_seconds=original.rolling_window_seconds,
            minimum_sample_count=original.minimum_sample_count,
            severity=original.severity,
            environment="PRODUCTION",
            effective_from=datetime.now(UTC),
            rationale="A deliberate test revision preserving immutable history.",
            created_by="test-operator",
        )
    )
    session.refresh(original)

    assert revised.revision == 2
    assert revised.active is True
    assert original.active is False
    assert original.fingerprint != revised.fingerprint

    evaluations = subject.evaluate_slos(datetime.now(UTC) + timedelta(seconds=1))
    vision = next(item for item in evaluations if item.policy_key == "vision.embedding.lab")
    production = next(item for item in evaluations if item.policy_key == "gateway.production")
    assert vision.state is SLOState.INSUFFICIENT_DATA
    assert vision.allowed_bad is None
    assert vision.remaining_bad is None
    assert production.allowed_bad is not None


def test_slo_evaluation_tracks_population_latency_and_burn_rates(session: Session) -> None:
    subject = service(session)
    subject.ensure_defaults()
    now = datetime.now(UTC)
    for index in range(20):
        completed = index != 19
        session.add(
            GatewayRequest(
                request_id=uuid.uuid4(),
                capability_key="rag.embedding",
                capability_version=1,
                status="completed" if completed else "failed",
                priority="interactive",
                input_sha256=f"{index:064x}",
                input_count=1,
                total_latency_ms=1500.0 if index == 18 else (100.0 if completed else None),
                failure_code=None if completed else "RUNTIME_FAILED",
                decision_evidence={"source": "m14-test"},
                created_at=now - timedelta(seconds=index),
                finished_at=now - timedelta(seconds=index) if completed else None,
            )
        )
    session.commit()

    results = subject.evaluate_slos(now + timedelta(seconds=1))
    success = next(
        item for item in results if item.policy_key == "rag.embedding.production.success"
    )
    latency = next(
        item for item in results if item.policy_key == "rag.embedding.production.latency"
    )

    assert success.sample_count == 20
    assert success.bad_count == 1
    assert success.state is SLOState.BREACHED
    assert success.long_burn_rate == pytest.approx(5.0)
    assert latency.sample_count == 19
    assert latency.bad_count == 1
    assert latency.observed_value == pytest.approx(240.0)
    assert latency.short_burn_rate == pytest.approx((1 / 19) / 0.05)


def test_slo_window_at_risk_stale_and_budget_recovery(session: Session) -> None:
    subject = service(session)
    subject.ensure_defaults()
    now = datetime.now(UTC)
    node = production_node(session, liveness="online")
    node.last_heartbeat_at = now - timedelta(seconds=100)
    for index in range(100):
        completed = index < 98
        session.add(
            GatewayRequest(
                request_id=uuid.uuid4(),
                capability_key="other.capability",
                capability_version=1,
                status="completed" if completed else "failed",
                priority="interactive",
                input_sha256=f"{index + 1000:064x}",
                input_count=1,
                total_latency_ms=100 if completed else None,
                failure_code=None if completed else "CONTROLLED_FAILURE",
                decision_evidence={},
                created_at=now - timedelta(minutes=index),
            )
        )
    session.add(
        GatewayRequest(
            request_id=uuid.uuid4(),
            capability_key="other.capability",
            capability_version=1,
            status="failed",
            priority="interactive",
            input_sha256="f" * 64,
            input_count=1,
            failure_code="OLD_FAILURE",
            decision_evidence={},
            created_at=now - timedelta(days=2),
        )
    )
    session.commit()

    first = subject.evaluate_slos(now)
    gateway = next(item for item in first if item.policy_key == "gateway.production")
    freshness = next(item for item in first if item.policy_key == "gpu_node.production.freshness")
    assert gateway.state is SLOState.AT_RISK
    assert gateway.sample_count == 100
    assert gateway.bad_count == 2
    assert gateway.remaining_bad == 0
    assert gateway.long_burn_rate == pytest.approx(2.0)
    assert freshness.state is SLOState.STALE

    recovered_at = now + timedelta(hours=25)
    node.last_heartbeat_at = recovered_at
    for index in range(20):
        session.add(
            GatewayRequest(
                request_id=uuid.uuid4(),
                capability_key="other.capability",
                capability_version=1,
                status="completed",
                priority="interactive",
                input_sha256=f"{index + 2000:064x}",
                input_count=1,
                total_latency_ms=100,
                decision_evidence={},
                created_at=recovered_at - timedelta(seconds=index),
            )
        )
    session.commit()
    recovered = subject.evaluate_slos(recovered_at)
    gateway_recovered = next(item for item in recovered if item.policy_key == "gateway.production")
    assert gateway_recovered.state is SLOState.HEALTHY
    assert gateway_recovered.bad_count == 0
    assert gateway_recovered.remaining_bad == pytest.approx(0.2)


def test_alert_pending_firing_acknowledged_resolved_and_correlated(session: Session) -> None:
    subject = service(session)
    subject.ensure_defaults()
    node = production_node(session)
    session.commit()
    start = datetime.now(UTC) - timedelta(seconds=30)

    first = next(
        item for item in subject.evaluate_alerts(start) if item.alert_type == "NODE_OFFLINE"
    )
    assert first.state is AlertState.PENDING
    firing = next(
        item
        for item in subject.evaluate_alerts(start + timedelta(seconds=11))
        if item.id == first.id
    )
    assert firing.state is AlertState.FIRING
    assert firing.occurrence_count == 2
    assert session.scalar(select(func.count()).select_from(OperationalIncident)) == 1

    acknowledged = subject.acknowledge(
        firing.id, AlertAction(actor="on-call", reason="Investigating node connectivity")
    )
    assert acknowledged.state is AlertState.ACKNOWLEDGED
    node.liveness_state = "online"
    session.commit()
    resolved = next(
        item
        for item in subject.evaluate_alerts(datetime.now(UTC) + timedelta(seconds=1))
        if item.id == first.id
    )
    assert resolved.state is AlertState.RESOLVED
    history = list(
        session.scalars(
            select(AlertHistoryEvent)
            .where(AlertHistoryEvent.alert_id == first.id)
            .order_by(AlertHistoryEvent.occurred_at)
        )
    )
    assert [item.to_state for item in history] == [
        "PENDING",
        "FIRING",
        "ACKNOWLEDGED",
        "RESOLVED",
    ]
    incident = session.scalar(select(OperationalIncident))
    assert incident is not None and incident.state == "RESOLVED"


def test_capability_rules_separate_lab_from_production(session: Session) -> None:
    subject = service(session)
    subject.ensure_defaults()
    production_rule = session.scalar(
        select(AlertRuleRevision).where(AlertRuleRevision.key == "capability.unavailable")
    )
    lab_rule = session.scalar(
        select(AlertRuleRevision).where(AlertRuleRevision.key == "capability.lab-unavailable")
    )

    assert production_rule is not None
    assert production_rule.condition == {"production_only": True}
    assert lab_rule is not None
    assert lab_rule.condition == {"lab_only": True}


def test_overview_reports_observability_persistence_degradation(session: Session) -> None:
    subject = service(session)
    subject.ensure_defaults()
    metrics.gauge("modelforge_observability_degraded", {}, 1)
    try:
        assert subject.overview().status == "OBSERVABILITY_DEGRADED"
    finally:
        metrics.gauge("modelforge_observability_degraded", {}, 0)


def test_maintenance_window_suppresses_matching_alert(session: Session) -> None:
    subject = service(session)
    subject.ensure_defaults()
    node = production_node(session)
    session.commit()
    now = datetime.now(UTC)
    subject.create_maintenance_window(
        MaintenanceWindowCreate(
            name="GPU Node maintenance rehearsal",
            starts_at=now - timedelta(minutes=1),
            ends_at=now + timedelta(minutes=10),
            matcher={"rule_key": "node.offline", "subject_ref": str(node.id)},
            reason="M14 deterministic suppression verification",
            created_by="test-operator",
        )
    )

    alert = next(item for item in subject.evaluate_alerts(now) if item.alert_type == "NODE_OFFLINE")

    assert alert.state is AlertState.SUPPRESSED
    assert alert.suppressed_until is not None
    assert alert.suppressed_until.replace(tzinfo=UTC) == now + timedelta(minutes=10)
    assert session.scalar(select(func.count()).select_from(OperationalIncident)) == 0


def test_alert_cooldown_reuses_fingerprint_and_storage_threshold_fires(
    session: Session,
) -> None:
    subject = service(session)
    subject.ensure_defaults()
    node = production_node(session)
    storage = StorageRoot(
        compute_node_id=node.id,
        name="models",
        purpose="model_artifacts",
        path="D:/models",
        status="ready",
        writable=True,
        capacity_bytes=100 * 1024**3,
        free_bytes=12 * 1024**3,
        reserve_bytes=10 * 1024**3,
        reserve_percent=10,
        validation_details={},
    )
    session.add(storage)
    session.commit()
    start = datetime.now(UTC) - timedelta(seconds=30)
    first = subject.evaluate_alerts(start)
    node_alert = next(item for item in first if item.alert_type == "NODE_OFFLINE")
    storage_alert = next(item for item in first if item.alert_type == "STORAGE_LOW")
    assert storage_alert.state is AlertState.FIRING
    firing = next(
        item
        for item in subject.evaluate_alerts(start + timedelta(seconds=11))
        if item.id == node_alert.id
    )
    assert firing.state is AlertState.FIRING
    node.liveness_state = "online"
    storage.free_bytes = 90 * 1024**3
    session.commit()
    resolved = next(
        item
        for item in subject.evaluate_alerts(start + timedelta(seconds=20))
        if item.id == node_alert.id
    )
    node.liveness_state = "offline"
    session.commit()
    during_cooldown = next(
        item
        for item in subject.evaluate_alerts(start + timedelta(seconds=30))
        if item.id == node_alert.id
    )
    assert during_cooldown.state is AlertState.RESOLVED
    recurred = next(
        item
        for item in subject.evaluate_alerts(start + timedelta(seconds=321))
        if item.id == node_alert.id
    )
    assert recurred.state is AlertState.PENDING
    assert recurred.fingerprint == resolved.fingerprint


def test_incident_groups_evidenced_downstream_only(session: Session) -> None:
    subject = service(session)
    subject.ensure_defaults()
    node = production_node(session)
    session.commit()
    now = datetime.now(UTC) - timedelta(seconds=20)
    subject.evaluate_alerts(now)
    node_alert = next(
        item
        for item in subject.evaluate_alerts(now + timedelta(seconds=11))
        if item.alert_type == "NODE_OFFLINE"
    )
    capability_rule = session.scalar(
        select(AlertRuleRevision).where(AlertRuleRevision.key == "capability.unavailable")
    )
    assert capability_rule is not None
    downstream = OperationalAlert(
        rule_id=capability_rule.id,
        fingerprint="d" * 64,
        alert_type="CAPABILITY_UNAVAILABLE",
        severity="CRITICAL",
        state="FIRING",
        source="capability_deployment",
        subject_type="capability",
        subject_ref="rag.embedding@1",
        summary="Stable capability cannot serve",
        details={},
        first_seen_at=now,
        last_seen_at=now,
        firing_at=now,
        occurrence_count=1,
    )
    unrelated = OperationalAlert(
        rule_id=capability_rule.id,
        fingerprint="e" * 64,
        alert_type="RUNTIME_CRASH_LOOP",
        severity="CRITICAL",
        state="FIRING",
        source="serving_jobs",
        subject_type="service",
        subject_ref="runtime",
        summary="Runtime failures",
        details={},
        first_seen_at=now,
        last_seen_at=now,
        firing_at=now,
        occurrence_count=1,
    )
    session.add_all((downstream, unrelated))
    session.flush()
    subject._correlate_incident(downstream, {"details": {"node_id": str(node.id)}}, now)
    subject._correlate_incident(unrelated, {"details": {}}, now)
    session.commit()
    root = session.get(OperationalAlert, node_alert.id)
    assert root is not None
    assert downstream.incident_id == root.incident_id
    assert unrelated.incident_id is None
    assert session.scalar(select(func.count()).select_from(OperationalIncident)) == 1


def test_capacity_trends_forecast_refusal_and_retention(session: Session) -> None:
    subject = service(session)
    node = production_node(session, liveness="online")
    accelerator = Accelerator(
        compute_node_id=node.id,
        device_index=0,
        device_uuid="GPU-M14",
        name="Test GPU",
        total_vram_bytes=16 * 1024**3,
        status="active",
        capabilities={},
    )
    session.add(accelerator)
    session.flush()
    telemetry = AcceleratorTelemetryLatest(
        accelerator_id=accelerator.id,
        used_vram_bytes=4 * 1024**3,
        free_vram_bytes=12 * 1024**3,
        availability={},
        observed_at=datetime.now(UTC),
        received_at=datetime.now(UTC),
    )
    storage = StorageVolumeState(
        compute_node_id=node.id,
        purpose="model_artifacts",
        path="D:/models",
        total_bytes=2 * 1024**4,
        used_bytes=1024**4,
        free_bytes=1024**4,
        availability={},
        observed_at=datetime.now(UTC),
        received_at=datetime.now(UTC),
    )
    session.add_all((telemetry, storage))
    session.commit()
    base = datetime.now(UTC) - timedelta(hours=2)

    for index in range(12):
        observed = base + timedelta(minutes=index * 10)
        telemetry.used_vram_bytes = (4 * 1024**3) + index * 1024**2
        telemetry.received_at = observed
        telemetry.observed_at = observed
        storage.free_bytes = (1024**4) - index * 1024**3
        storage.observed_at = observed
        storage.received_at = observed
        session.commit()
        subject.collect_capacity(observed)

    trend = subject.capacity_trend(node.id, hours=4)
    assert trend.status == "AVAILABLE"
    assert trend.sample_count == 12
    assert trend.metrics["gpu_schedulable_p95"] is not None
    assert trend.metrics["external_gpu_max"] is not None
    assert trend.forecast["status"] == "AVAILABLE"
    assert trend.forecast["storage_change_bytes_per_day"] < 0

    snapshots = list(
        session.scalars(select(CapacitySnapshot).order_by(CapacitySnapshot.observed_at))
    )
    for item in snapshots[1:]:
        session.delete(item)
    session.commit()
    refused = subject.capacity_trend(node.id, hours=4)
    assert refused.status == "INSUFFICIENT_DATA"
    assert refused.forecast == {"status": "INSUFFICIENT_DATA"}

    old = snapshots[0]
    old.observed_at = datetime.now(UTC) - timedelta(days=8)
    session.add(old)
    session.commit()
    retention = subject.apply_retention(datetime.now(UTC))
    assert retention == {
        "aggregates_created": 1,
        "raw_deleted": 1,
        "aggregates_deleted": 0,
    }
    assert session.scalar(select(func.count()).select_from(CapacityAggregate)) == 1
    assert session.scalar(select(func.count()).select_from(CapacitySnapshot)) == 0