app/observability/contracts.py

"""Canonical observation and evaluation contracts for RAG telemetry.

Every later observability task (OTel boundary, RAG instrumentation, benchmark
harness, Control Room wiring, SigNoz dashboards, the final write-up) imports
these shapes instead of redefining its own. The goal is the one stated in the
plan preamble: "Every value has one name and meaning across local
observations, spans, metrics, Control Room, SigNoz, and the final report."

This module defines shapes only -> it emits nothing. No span, metric, log, or
file write happens here.
"""

from __future__ import annotations

from typing import Any, Literal

from pydantic import BaseModel, ConfigDict, Field

# --- Telemetry mode -------------------------------------------------

# "disabled": no OTel emission at all.
# "local": emit locally (local_store JSONL / Control Room) without exporting
#          to a remote collector.
# "full": emit locally AND export via OTLP to SigNoz.
TelemetryMode = Literal["disabled", "local", "full"]


# --- Retrieval outcome -------------------------------------------------

# Deliberately distinct from a plain empty list: "success_empty" means the
# retrieval pipeline ran correctly and genuinely found nothing, while
# "error_fallback" means the pipeline itself failed and the caller fell back
# to an empty/degraded result. Collapsing these into one signal was the
# whole reason retrieval failures used to be invisible before instrumentation.
RetrievalStatus = Literal["success", "success_empty", "error_fallback"]


class RetrievalOutcome(BaseModel):
    """The result of one retrieval attempt, kept distinct from a plain list.

    Product code and instrumentation both consume this: product code gets a
    real status even in fallback paths, and observability code gets a
    ready-made `retrieval_status` / `empty_result` / `retrieval_error` triple
    per the canonical "Retrieval-behavior measures" contract.
    """

    model_config = ConfigDict(extra="forbid")

    status: RetrievalStatus
    candidates: list[dict[str, Any]] = Field(default_factory=list)
    empty_result: bool = False
    retrieval_error: bool = False
    error_type: str | None = None

    @classmethod
    def success(cls, candidates: list[dict[str, Any]]) -> "RetrievalOutcome":
        return cls(
            status="success" if candidates else "success_empty",
            candidates=candidates,
            empty_result=not candidates,
        )

    @classmethod
    def failure(cls, error_type: str) -> "RetrievalOutcome":
        return cls(status="error_fallback", retrieval_error=True, error_type=error_type)


# --- Metric-safe attribute filtering -------------------------------------------------

# Exactly the "Metric-safe dimensions" list from the plan preamble. Never add
# run/query/trace/project/document/operation IDs here -> those stay trace and
# log attributes only, per the global constraint that plain project IDs are
# allowed as trace/log attributes but must never become metric dimensions.
METRIC_ATTRIBUTE_KEYS: frozenset[str] = frozenset(
    {
        "pipeline.version",
        "query.category",
        "cache.state",
        "consumer",
        "operation",
        "subsystem",
        "status",
        "service.name",
        "evaluation_run",
        # Single bounded operator-controlled boolean. This lets canonical
        # product cohorts exclude controlled diagnostic signals without using
        # a high-cardinality run or request identifier.
        "diagnostic.smoke",
    }
)


def metric_attributes(attributes: dict[str, Any]) -> dict[str, Any]:
    """Narrow an arbitrary attribute dict down to metric-safe dimensions.

    Keeps only keys in `METRIC_ATTRIBUTE_KEYS` whose value is a plain scalar
    (str/int/float/bool). Non-scalar values (lists, dicts, None) are dropped
    even for an otherwise-safe key, since a metric dimension must be a single
    bounded value, not a structure that can blow up cardinality on its own.
    Does not mutate the input.
    """
    return {
        key: value
        for key, value in attributes.items()
        if key in METRIC_ATTRIBUTE_KEYS and isinstance(value, (str, int, float, bool))
    }


# --- Experiment identity -------------------------------------------------


class ExperimentIdentity(BaseModel):
    """The canonical experiment-identity fields from the plan preamble.

    Ordinary product traffic leaves these unset; labelled evaluation runs
    (the benchmark harness in a later task) populate them so every trace,
    metric, log, and local record for one run can be joined back together.
    Field names are snake_case; the dotted preamble names they mirror are
    noted alongside each field.
    """

    model_config = ConfigDict(extra="forbid")

    experiment_id: str | None = None  # experiment.id
    run_id: str | None = None  # experiment.run_id
    pipeline_version: str | None = None  # pipeline.version
    query_set_version: str | None = None  # query_set.version
    corpus_version: str | None = None  # corpus.version
    query_id: str | None = None  # query.id
    query_category: str | None = None  # query.category
    repetition: int | None = None  # repetition
    cache_state: str | None = None  # cache.state
    telemetry_mode: TelemetryMode | None = None  # telemetry.mode
    cold_start: bool | None = None
    git_commit: str | None = None  # git.commit
    embedding_model: str | None = None  # embedding.model
    cerebras_model: str | None = None  # cerebras.model
    chroma_version: str | None = None  # chroma.version
    cognee_version: str | None = None  # cognee.version


# --- Operation observation -------------------------------------------------


class OperationObservation(BaseModel):
    """Compact, per-operation local observation.

    This is the shape the (not-yet-built) single operation helper produces
    for Control Room and experiment evidence: "latest stage durations,
    counts, statuses, and trace ID" per the preamble's signal-responsibility
    table. It carries no prompts, full responses, PDF text, chunks,
    annotation contents, filesystem paths, or raw document IDs -> only
    durations, counts, statuses, and identity.

    `status` is intentionally a plain string rather than a narrow Literal:
    it must accommodate both ordinary per-operation statuses (e.g. "success",
    "error") and the canonical missing-data sentinels (`not_instrumented`,
    `not_applicable`, `not_observed`, `degraded`) -> a numeric duration must
    stay `None` under those sentinels, never silently become 0.
    """

    model_config = ConfigDict(extra="forbid")

    operation: str
    status: str
    subsystem: str | None = None
    consumer: str | None = None
    duration_ms: float | None = None
    stage_durations_ms: dict[str, float] = Field(default_factory=dict)
    counts: dict[str, int | float] = Field(default_factory=dict)
    trace_id: str | None = None
    experiment: ExperimentIdentity | None = None
    retrieval: RetrievalOutcome | None = None
    evaluation_run: bool = False
    attributes: dict[str, Any] = Field(default_factory=dict)