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

Three drill-down tools on top of the L1 group / table inventory:

- list_monitor_events(table_group_id, table_name, monitor_type, monitor_id?,
  include_predictions?, limit?, page?) — per-table event history scoped to one
  monitor type. monitor_id is required when monitor_type="metric" (Metric is
  the only multi-instance type — singletons reject monitor_id with a clear
  pointer). Per-type column shapes:
    * Volume:    Time | Status | Row count | Lower bound | Upper bound
    * Freshness: Time | Status | Update detected | Detail
                 (parsed from the SQL template's structured message)
    * Schema:    Time | Status | Table change | Columns added | Columns dropped | Columns modified
    * Metric:    Time | Status | Value | Lower bound | Upper bound
                 (metric name in the heading, not as a column)
  include_predictions=True appends a separate `## Forecast` section listing
  future timestamps with predicted bounds — never interleaved into the
  historical events. Schema short-circuits to "not applicable"; non-
  Prediction-Model monitors render a "not available" note.

- list_monitors(table_group_id, table_name) — configured monitors for a
  table. Per row: monitor_id, type (Title Case, reuses _MONITOR_LABEL),
  metric_name, threshold mode (derived from history_calculation:
  "PREDICT" → Prediction; non-empty otherwise and not Freshness →
  Historical; empty → Static), bounds, and the metric expression for
  Metric monitors. Sensitivity is surfaced as a top-level "Prediction
  model sensitivity" field.

- list_monitor_schema_changes(table_group_id, table_name?, since?,
  limit?, page?) — newest-first audit log of column-level schema events
  (added / dropped / modified), independent of monitor-run results.

All three follow the L1 contract: gated on view, resolve_monitored_table_group
returns the literal "This table group is not monitored." when the group has
no linked monitor suite, and never expose internal test_type codes,
test_definition_id, or test_suite_id on the surface.

Backing model methods:

- DataStructureLog ORM wraps the existing data_structure_log audit table
  (no ORM mapping until now) and exposes list_for_table_group(*, table_name,
  since, until, page, limit). The dashboard's get_data_structure_logs
  delegates here.
- TestDefinition.list_monitor_configs_for_table returns MonitorConfig with
  threshold_mode derived per row.
- TestDefinition.get_singleton_monitor(suite, table, type) for the non-
  metric forecast lookup path.
- forecast_points_from_prediction(prediction, sensitivity) — standalone
  helper that reads forecast rows from the prediction JSONB using epoch-ms
  keys with numeric comparison (matches the dashboard's reader).
- TestResult.list_monitor_events_for_table runs the dashboard's CTE under
  the ORM with optional monitor_type filtering. ORDER BY includes
  results.id NULLS LAST, active_runs.id as a stable tiebreaker so the
  Python-side pagination doesn't duplicate or skip rows on test_time ties.
- TestResult.list_metric_monitor_events — separate, simpler query path
  for Metric (no run-by-type CROSS JOIN, no synthesized pending rows).

The dashboard's per-type events transform stays in monitors_dashboard.py
because its shape is bespoke to the chart payload; centralizing just the
SQL would still require a non-trivial adapter on the dashboard side.

monitor_id joins the Followable IDs table as an alias for the underlying
test_definition.id UUID — the abstraction stays stable if monitors move
to a dedicated table later.

Status helpers (_summary_status, _format_monitor_cell, _format_schema_cell,
_event_status) return Title Case ("Error", "Pending", "Training",
"Anomaly", "Ok"), matching the rest of the MCP layer. parse_monitor_type
accepts either case so values round-trip.

_parse_kv_pairs splits input_parameters on ";" (matches the writer side
at execute_tests_query.py:321 and the dashboard's dict_from_kv).

Closes TG-1092

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: 2 people

testgen/common/models/data_structure_log.py

@@ -0,0 +1,93 @@
+from dataclasses import dataclass
+from datetime import date, 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,
+        *,
+        table_name: str | None = None,
+        since: date | datetime | None = None,
+        until: date | datetime | None = None,
+        page: int = 1,
+        limit: int | None = 20,
+    ) -> tuple[list[DataStructureLogEntry], int]:
+        """Paginated schema-change audit log for one table group, newest first.
+
+        Filters by ``table_name`` (exact match; the audit log stores names
+        case-sensitively as the source emitted them), ``since`` (lower-bound on
+        ``change_date``), and ``until`` (upper-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)
+        if table_name is not None:
+            query = query.where(cls.table_name == table_name)
+        if since is not None:
+            query = query.where(cls.change_date >= since)
+        if until is not None:
+            query = query.where(cls.change_date <= until)
+        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,96 @@ class TestDefinitionMinimal(EntityMinimal):
     test_name_short: str
 
 
+# Threshold-mode labels for monitor TestDefinitions — derived from which fields
+# on the definition are populated. See ``TestDefinition._derive_threshold_mode``.
+THRESHOLD_MODE_PREDICTION = "Prediction Model"
+THRESHOLD_MODE_HISTORICAL = "Historical Calculation"
+THRESHOLD_MODE_STATIC = "Static"
+THRESHOLD_MODE_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: str
+    threshold_lower: str | None
+    threshold_upper: str | None
+    custom_query: str | None
+
+
 class QueryString(TypeDecorator):
     impl = String
     cache_ok = True
@@ -557,6 +648,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[str, 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 THRESHOLD_MODE_NONE, None, None
+        if td.history_calculation == "PREDICT":
+            return THRESHOLD_MODE_PREDICTION, None, None
+        if td.history_calculation and td.test_type != MonitorType.FRESHNESS.value:
+            return THRESHOLD_MODE_HISTORICAL, td.history_calculation, td.history_calculation_upper
+        if td.test_type == MonitorType.FRESHNESS.value:
+            return THRESHOLD_MODE_STATIC, None, td.upper_tolerance
+        return THRESHOLD_MODE_STATIC, td.lower_tolerance, td.upper_tolerance
+
     @classmethod
     def select_page(
         cls,