"""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)