tests/observability/test_observability_summary.py

"""Status-vocabulary and aggregation tests for the Control Room observability
projection (`app.services.observability_summary.observability_summary`).

Uses real `bootstrap.initialize_observability(...)` + real
`LocalObservationStore` instances against a `tmp_path` artifact root -- the
same isolation pattern `tests/observability/test_rag_spans.py` and
`test_operation.py` already use -- rather than mocking the observability
layer.
"""

from __future__ import annotations

import pytest

from app.observability import bootstrap
from app.observability.config import ObservabilityConfig
from app.observability.operation import observe_operation
from app.services.observability_summary import observability_summary


@pytest.fixture(autouse=True)
def _shutdown_after():
    bootstrap.reset_suppressed_failures()
    yield
    bootstrap.shutdown_observability(timeout_seconds=2.0)


def _config(tmp_path, mode: str) -> ObservabilityConfig:
    return ObservabilityConfig(
        enabled=mode != "disabled",
        mode=mode,  # type: ignore[arg-type]
        otlp_endpoint="http://127.0.0.1:4318",
        signoz_ui_url="http://127.0.0.1:8080",
        artifact_root=tmp_path,
    )


# --- not_instrumented ---------------------------------------------------


def test_not_instrumented_when_mode_disabled(tmp_path):
    bootstrap.initialize_observability(config=_config(tmp_path, "disabled"))

    summary = observability_summary()

    assert summary["status"] == "not_instrumented"
    assert summary["latest_trace_id"] is None
    assert summary["pipeline_version"] is None
    assert summary["stages"] == {}
    assert summary["retrieval"] == {"success_count": 0, "empty_count": 0, "error_count": 0}
    assert summary["exporter"] == {"mode": "disabled", "suppressed_failures": 0}
    assert summary["signoz_url"] is None


# --- not_observed ---------------------------------------------------


def test_not_observed_when_store_empty(tmp_path):
    bootstrap.initialize_observability(config=_config(tmp_path, "local"))

    summary = observability_summary()

    assert summary["status"] == "not_observed"
    assert summary["latest_trace_id"] is None
    assert summary["stages"] == {}
    assert summary["retrieval"] == {"success_count": 0, "empty_count": 0, "error_count": 0}
    assert summary["exporter"]["mode"] == "local"


def test_not_observed_takes_priority_over_signoz_url_in_full_mode(tmp_path):
    bootstrap.initialize_observability(config=_config(tmp_path, "full"))

    summary = observability_summary()

    assert summary["status"] == "not_observed"
    # full mode always surfaces the SigNoz link, even with nothing observed yet.
    assert summary["signoz_url"] == "http://127.0.0.1:8080"


# --- degraded ---------------------------------------------------


def test_degraded_when_suppressed_failures_recorded_even_with_data(tmp_path):
    bootstrap.initialize_observability(config=_config(tmp_path, "local"))
    with observe_operation("rag.retrieval"):
        pass
    bootstrap.note_suppressed_failure()

    summary = observability_summary()

    assert summary["status"] == "degraded"
    assert summary["exporter"]["suppressed_failures"] >= 1


# --- observed ---------------------------------------------------


def test_observed_with_stage_percentiles_and_retrieval_counts(tmp_path):
    from app.observability.contracts import RetrievalOutcome

    bootstrap.initialize_observability(config=_config(tmp_path, "local"))

    # Three vector_search rows with distinct durations to exercise p50/p95.
    for _ in range(3):
        with observe_operation("chroma.vector_search"):
            pass

    with observe_operation("rag.retrieval") as op:
        op.set_retrieval(RetrievalOutcome.success([{"text": "a"}]))
    with observe_operation("rag.retrieval") as op:
        op.set_retrieval(RetrievalOutcome.success([]))
    with observe_operation("rag.retrieval") as op:
        op.set_retrieval(RetrievalOutcome.failure("vector_search_failed"))
        op.mark_error("vector_search_failed")

    summary = observability_summary()

    assert summary["status"] == "observed"
    assert summary["latest_trace_id"] is not None
    assert "vector_search_ms" in summary["stages"]
    assert summary["stages"]["vector_search_ms"]["count"] == 3
    assert summary["stages"]["vector_search_ms"]["p50"] >= 0
    assert summary["stages"]["vector_search_ms"]["p95"] >= summary["stages"]["vector_search_ms"]["p50"]
    assert "retrieval_ms" in summary["stages"]
    assert summary["stages"]["retrieval_ms"]["count"] == 3
    assert summary["retrieval"] == {"success_count": 1, "empty_count": 1, "error_count": 1}
    assert summary["exporter"] == {"mode": "local", "suppressed_failures": 0}


def test_missing_duration_never_counted_as_zero(tmp_path):
    """A row recorded via a store that never got a real duration (e.g. an
    injected store bypassing the timing boundary) must not pollute p50/p95.

    Exercised here by directly appending a row through the store with
    `duration_ms=None`, the way a degraded/failed timing path could produce
    one, and confirming it contributes zero to the stage's count.
    """
    from app.observability.contracts import OperationObservation
    from app.observability.local_store import LocalObservationStore

    bootstrap.initialize_observability(config=_config(tmp_path, "local"))
    store = bootstrap.get_state().local_store
    assert isinstance(store, LocalObservationStore)
    store.record(
        OperationObservation(operation="chroma.vector_search", status="success", duration_ms=None)
    )

    summary = observability_summary()

    assert summary["status"] == "observed"
    assert summary["stages"] == {}


def test_pipeline_version_taken_from_latest_experiment(tmp_path):
    from app.observability.contracts import ExperimentIdentity

    bootstrap.initialize_observability(config=_config(tmp_path, "local"))
    with observe_operation("rag.retrieval") as op:
        op.set_experiment(ExperimentIdentity(pipeline_version="rag-naive-v1"))
    with observe_operation("rag.retrieval"):
        pass  # no experiment identity -- ordinary traffic

    summary = observability_summary()

    # Most recent row wins; the plain ordinary-traffic row carries no
    # experiment identity, so the last *tagged* row's version is used.
    assert summary["pipeline_version"] == "rag-naive-v1"


def test_signoz_url_only_present_in_full_mode(tmp_path):
    bootstrap.initialize_observability(config=_config(tmp_path, "full"))
    with observe_operation("rag.retrieval"):
        pass

    summary = observability_summary()

    assert summary["signoz_url"] == "http://127.0.0.1:8080"
    assert summary["exporter"]["mode"] == "full"