Merge branch 'feat/TG-1092-mcp-monitors-l2' into 'enterprise'

feat(mcp): add monitor L2 read tools — per-table events, configs, and schema-change audit (TG-1092)

See merge request dkinternal/testgen/dataops-testgen!563

Author: ci bot

testgen/common/models/data_structure_log.py

@@ -0,0 +1,85 @@
+from dataclasses import dataclass
+from datetime import datetime
+from uuid import UUID, uuid4
+
+from sqlalchemy import Column, String, asc, desc, select
+from sqlalchemy.dialects import postgresql
+
+from testgen.common.models import get_current_session
+from testgen.common.models.entity import Entity, EntityMinimal
+
+# Schema-change codes stored in ``data_structure_log.change``. ``M`` only ever
+# appears on column-level rows (column-name set, old/new data type populated).
+SCHEMA_CHANGE_ADDED = "A"
+SCHEMA_CHANGE_DROPPED = "D"
+SCHEMA_CHANGE_MODIFIED = "M"
+
+
+@dataclass
+class DataStructureLogEntry(EntityMinimal):
+    """One schema-change event for a table or column in a table group.
+
+    ``column_name`` is ``None`` for table-level events (add / drop of an entire
+    table); set for column-level events. ``change`` is the stored single-letter
+    code (``A`` / ``D`` / ``M``); callers translate to user-facing words.
+    """
+    log_id: UUID
+    table_groups_id: UUID
+    table_name: str
+    column_name: str | None
+    change_date: datetime
+    change: str
+    old_data_type: str | None
+    new_data_type: str | None
+
+
+class DataStructureLog(Entity):
+    __tablename__ = "data_structure_log"
+
+    log_id: UUID = Column(postgresql.UUID(as_uuid=True), primary_key=True, default=uuid4)
+    table_groups_id: UUID = Column(postgresql.UUID(as_uuid=True))
+    table_id: UUID = Column(postgresql.UUID(as_uuid=True))
+    column_id: UUID = Column(postgresql.UUID(as_uuid=True))
+    table_name: str = Column(String)
+    column_name: str = Column(String)
+    change_date: datetime = Column(postgresql.TIMESTAMP)
+    change: str = Column(String)
+    old_data_type: str = Column(String)
+    new_data_type: str = Column(String)
+
+    _get_by = "log_id"
+    _default_order_by = (desc(change_date), asc(table_name), asc(column_name))
+
+    @classmethod
+    def list_for_table_group(
+        cls,
+        table_group_id: str | UUID,
+        *clauses,
+        page: int = 1,
+        limit: int | None = 20,
+    ) -> tuple[list[DataStructureLogEntry], int]:
+        """Paginated schema-change audit log for one table group, newest first.
+
+        Caller-supplied ``*clauses`` are WHERE expressions on this model — e.g.
+        ``cls.table_name == name`` (exact match; the audit log stores names
+        case-sensitively as the source emitted them) or a ``cls.change_date``
+        bound. ``limit=None`` skips pagination — the caller gets every matching
+        row in one shot.
+        """
+        query = select(
+            cls.log_id,
+            cls.table_groups_id,
+            cls.table_name,
+            cls.column_name,
+            cls.change_date,
+            cls.change,
+            cls.old_data_type,
+            cls.new_data_type,
+        ).where(cls.table_groups_id == table_group_id, *clauses)
+        query = query.order_by(*cls._default_order_by)
+
+        if limit is None:
+            rows = get_current_session().execute(query).mappings().all()
+            entries = [DataStructureLogEntry(**row) for row in rows]
+            return entries, len(entries)
+        return cls._paginate(query, page=page, limit=limit, data_class=DataStructureLogEntry)

testgen/common/models/test_definition.py

@@ -27,6 +27,7 @@
 from sqlalchemy.orm import InstrumentedAttribute
 from sqlalchemy.sql.expression import case, literal
 
+from testgen.common.enums import MonitorType
 from testgen.common.models import Base, get_current_session
 from testgen.common.models.custom_types import NullIfEmptyString, YNString, ZeroIfEmptyInteger
 from testgen.common.models.entity import Entity, EntityMinimal
@@ -270,6 +271,97 @@ class TestDefinitionMinimal(EntityMinimal):
     test_name_short: str
 
 
+class ThresholdMode(StrEnum):
+    """How a monitor's bounds are determined — derived from which fields on the
+    definition are populated. See ``TestDefinition._derive_threshold_mode``."""
+    PREDICTION = "Prediction Model"
+    HISTORICAL = "Historical Calculation"
+    STATIC = "Static"
+    NONE = "N/A"
+
+
+@dataclass
+class MonitorForecastPoint(EntityMinimal):
+    """One future forecast point read from a Prediction-Model monitor's stored
+    ``prediction`` JSONB. Each point is a ``(test_time, lower_bound, upper_bound)``
+    triple at a specific upcoming timestamp; collectively they extend the
+    historical event series forward under the suite's active prediction
+    sensitivity. Surfaced as a separate forecast section on
+    ``list_monitor_events``, never as an event."""
+    test_time: datetime
+    lower_bound: float | None
+    upper_bound: float | None
+
+
+def forecast_points_from_prediction(
+    prediction: dict | None,
+    sensitivity: str,
+) -> list[MonitorForecastPoint]:
+    """Extract forecast points for a sensitivity from a monitor's stored
+    ``prediction`` JSONB, sorted by time ascending (nearest future point
+    first).
+
+    The JSONB is keyed as ``"lower_tolerance|<sensitivity>" → {epoch_ms: value}``
+    and ``"upper_tolerance|<sensitivity>" → {epoch_ms: value}`` — matches the
+    format the dashboard reads at ``monitors_dashboard.py`` via
+    ``datetime.fromtimestamp(int(timestamp) / 1000.0, UTC)``. Returns ``[]``
+    when the monitor isn't in Prediction Model mode (no JSONB) or when the
+    sensitivity has no stored series. Standalone so it works against either
+    the ``TestDefinition`` ORM row or the ``TestDefinitionSummary`` dataclass
+    — both carry the same ``prediction`` field shape.
+    """
+    if not prediction:
+        return []
+    lower_series = prediction.get(f"lower_tolerance|{sensitivity}") or {}
+    upper_series = prediction.get(f"upper_tolerance|{sensitivity}") or {}
+    if not lower_series and not upper_series:
+        return []
+
+    all_keys = sorted(set(lower_series) | set(upper_series), key=int)
+    points: list[MonitorForecastPoint] = []
+    for k in all_keys:
+        ts = datetime.fromtimestamp(int(k) / 1000.0, UTC)
+        lower = lower_series.get(k)
+        upper = upper_series.get(k)
+        points.append(MonitorForecastPoint(
+            test_time=ts,
+            lower_bound=float(lower) if lower is not None else None,
+            upper_bound=float(upper) if upper is not None else None,
+        ))
+    return points
+
+
+@dataclass
+class MonitorConfig(EntityMinimal):
+    """One configured monitor — produced by ``TestDefinition.list_monitor_configs_for_table``.
+
+    ``threshold_mode`` is derived from the underlying definition's
+    ``history_calculation``: ``"PREDICT"`` flags Prediction Model; any other
+    non-empty value flags Historical Calculation (not available for Freshness);
+    empty falls through to Static — the default for Freshness, Volume, and
+    Metric. Schema_Drift is presence-only and reports N/A.
+
+    ``threshold_lower`` / ``threshold_upper`` carry the bounds active under
+    the current mode — static tolerances for Static, history-calc expressions
+    (e.g. ``"Minimum"`` / ``"Maximum"``) for Historical, ``None`` for
+    Prediction (the runtime bands live on each event, not the configuration).
+    For Freshness in Static mode only the upper bound applies.
+
+    ``metric_name`` is the user-defined name for a Metric monitor (stored on
+    the underlying definition's ``column_name`` column, but it is the metric's
+    name rather than a column reference). ``custom_query`` is the metric's SQL
+    expression. Both are only set for ``Metric_Trend``.
+    """
+    monitor_id: UUID
+    test_type: str
+    table_name: str
+    metric_name: str | None
+    threshold_mode: ThresholdMode
+    threshold_lower: str | None
+    threshold_upper: str | None
+    custom_query: str | None
+
+
 class QueryString(TypeDecorator):
     impl = String
     cache_ok = True
@@ -557,6 +649,95 @@ def list_for_suite(
         query = query.order_by(*cls._default_order_by)
         return cls._paginate(query, page=page, limit=limit, data_class=TestDefinitionSummary)
 
+    @classmethod
+    def get_singleton_monitor(
+        cls,
+        test_suite_id: str | UUID,
+        table_name: str,
+        test_type: str,
+    ) -> "TestDefinition | None":
+        """Return the single ``TestDefinition`` row for a singleton monitor
+        type (Freshness / Volume / Schema) on a given table. Metric is
+        multi-instance and should be looked up by ``id`` instead — this
+        helper would silently pick the first match. Returns ``None`` when no
+        monitor is configured."""
+        session = get_current_session()
+        return session.execute(
+            select(cls)
+            .where(cls.test_suite_id == test_suite_id)
+            .where(cls.table_name == table_name)
+            .where(cls.test_type == test_type)
+            .limit(1)
+        ).scalars().first()
+
+    @classmethod
+    def list_monitor_configs_for_table(
+        cls,
+        test_suite_id: str | UUID,
+        table_name: str,
+    ) -> list[MonitorConfig]:
+        """List configured monitors for a single table within a monitor suite.
+
+        Returns one entry per ``test_definition`` row whose ``test_type`` is one
+        of the four monitor types. Typically 3-4 rows per table; Metric_Trend
+        contributes one entry per metric, so a table may have more.
+        """
+        monitor_codes = [m.value for m in MonitorType]
+        defs = cls.select_where(
+            cls.test_suite_id == test_suite_id,
+            cls.table_name == table_name,
+            cls.test_type.in_(monitor_codes),
+        )
+        return [cls._build_monitor_config(td) for td in defs]
+
+    @classmethod
+    def _build_monitor_config(cls, td: "TestDefinition") -> MonitorConfig:
+        mode, threshold_lower, threshold_upper = cls._derive_threshold_mode(td)
+        return MonitorConfig(
+            monitor_id=td.id,
+            test_type=td.test_type,
+            table_name=td.table_name,
+            metric_name=td.column_name or None if td.test_type == MonitorType.METRIC.value else None,
+            threshold_mode=mode,
+            threshold_lower=threshold_lower,
+            threshold_upper=threshold_upper,
+            custom_query=td.custom_query if td.test_type == MonitorType.METRIC.value else None,
+        )
+
+    @classmethod
+    def _derive_threshold_mode(
+        cls, td: "TestDefinition",
+    ) -> tuple[ThresholdMode, str | None, str | None]:
+        """Pick a mode and the bounds tuple that applies under that mode.
+
+        Detection mirrors the UI form (``test_definition_form.js``): a
+        ``history_calculation`` of exactly ``"PREDICT"`` flags Prediction
+        mode; any other non-empty value flags Historical (not available for
+        Freshness); empty falls through to Static — the default for
+        Freshness / Volume / Metric. Schema never has thresholds.
+
+        Bounds returned per mode:
+
+        * Prediction: ``None, None``. Runtime bounds live in the per-run
+          prediction JSONB; they are not configuration and do not surface
+          here.
+        * Historical: ``(history_calculation, history_calculation_upper)`` —
+          stored as expressions like ``"Minimum"`` / ``"Maximum"`` that the
+          execution layer evaluates against the lookback window.
+        * Static for Freshness: ``(None, upper_tolerance)``. Only an upper
+          bound applies.
+        * Static (Volume / Metric): ``(lower_tolerance, upper_tolerance)``.
+        """
+        if td.test_type == MonitorType.SCHEMA.value:
+            return ThresholdMode.NONE, None, None
+        if td.history_calculation == "PREDICT":
+            return ThresholdMode.PREDICTION, None, None
+        if td.history_calculation and td.test_type != MonitorType.FRESHNESS.value:
+            return ThresholdMode.HISTORICAL, td.history_calculation, td.history_calculation_upper
+        if td.test_type == MonitorType.FRESHNESS.value:
+            return ThresholdMode.STATIC, None, td.upper_tolerance
+        return ThresholdMode.STATIC, td.lower_tolerance, td.upper_tolerance
+
     @classmethod
     def select_page(
         cls,