diff --git a/.github/workflows/test-warehouse.yml b/.github/workflows/test-warehouse.yml
index 675d65cc2..c167516bc 100644
--- a/.github/workflows/test-warehouse.yml
+++ b/.github/workflows/test-warehouse.yml
@@ -60,7 +60,7 @@ env:
 jobs:
   test:
     runs-on: ubuntu-latest
-    timeout-minutes: 60
+    timeout-minutes: 90
     permissions:
       contents: read
       # Mint an OIDC token to assume the shared elementary-oss AWS role.
diff --git a/README.md b/README.md
index 3be912b5f..8aef2480e 100644
--- a/README.md
+++ b/README.md
@@ -32,6 +32,8 @@ These tables are the backbone of any observability setup — enabling alerts, re
 **2. Elementary Tests**
 A suite of anomaly detection and data quality tests that run like native dbt tests — no separate tooling. Covers volume, freshness, column distributions, schema changes, and AI-powered validation for structured and unstructured data. → [See all tests](https://docs.elementary-data.com/data-tests/introduction)
 
+[Metric stability](https://docs.elementary-data.com/data-tests/metric-stability) detects restatements of settled historical aggregates within a configured observation window.
+
 ---
 
 ## Quickstart
diff --git a/integration_tests/tests/test_metric_stability.py b/integration_tests/tests/test_metric_stability.py
new file mode 100644
index 000000000..1fe162313
--- /dev/null
+++ b/integration_tests/tests/test_metric_stability.py
@@ -0,0 +1,447 @@
+import json
+from datetime import datetime, time, timedelta
+from itertools import pairwise
+from typing import Any, Dict, List, Optional
+
+import pytest
+from data_generator import DATE_FORMAT
+from dbt_project import DbtProject
+
+TIMESTAMP_COLUMN = "updated_at"
+VALUE_COLUMN = "amount"
+OTHER_VALUE_COLUMN = "other_amount"
+DBT_TEST_NAME = "elementary.metric_stability"
+
+BASE_AMOUNT = 100
+OTHER_BASE_AMOUNT = 500
+DAYS_OF_HISTORY = 6
+
+# A min_bucket_age of one day means buckets older than a day are checked, and the
+# derived observation window (twice the age) keeps them being measured, so a
+# bucket two days old is both settled and still under observation.
+SETTLED_DAYS_AGO = 2
+UNSETTLED_DAYS_AGO = 1
+
+BASE_ARGS: Dict[str, Any] = {
+    "columns": [VALUE_COLUMN],
+    "metrics": ["sum"],
+    "timestamp_column": TIMESTAMP_COLUMN,
+    "time_bucket": {"period": "day", "count": 1},
+    "min_bucket_age": {"count": 1, "period": "day"},
+}
+
+
+def _rows(
+    restatements: Optional[Dict[int, int]] = None,
+    other_restatements: Optional[Dict[int, int]] = None,
+) -> List[Dict[str, Any]]:
+    """One row per day, midday so it lands unambiguously inside a daily bucket."""
+    restatements = restatements or {}
+    other_restatements = other_restatements or {}
+    utc_today = datetime.utcnow().date()
+    rows = []
+    for days_ago in range(1, DAYS_OF_HISTORY + 1):
+        timestamp = datetime.combine(utc_today - timedelta(days=days_ago), time(12, 0))
+        rows.append(
+            {
+                TIMESTAMP_COLUMN: timestamp.strftime(DATE_FORMAT),
+                VALUE_COLUMN: restatements.get(days_ago, BASE_AMOUNT),
+                OTHER_VALUE_COLUMN: other_restatements.get(days_ago, OTHER_BASE_AMOUNT),
+            }
+        )
+    return rows
+
+
+def _run(dbt_project: DbtProject, test_id: str, data, **overrides) -> str:
+    result = dbt_project.test(
+        test_id,
+        DBT_TEST_NAME,
+        {**BASE_ARGS, **overrides},
+        data=data,
+        test_vars={"enable_elementary_test_materialization": True},
+    )
+    return result["status"]
+
+
+def _bucket_values(
+    dbt_project: DbtProject, test_id: str, column_name: str = VALUE_COLUMN
+) -> Dict[Any, List[float]]:
+    """Measured values per bucket, oldest measurement first.
+
+    Asserting on these rather than only on pass/fail means a wrong baseline or a
+    wrong sign in the comparison cannot slip through.
+    """
+    metrics = dbt_project.read_table(
+        "data_monitoring_metrics",
+        where=(
+            f"full_table_name LIKE '%{test_id.upper()}' "
+            f"and metric_name = 'sum' "
+            f"and lower(column_name) = '{column_name}'"
+        ),
+    )
+    by_bucket: Dict[Any, List[Any]] = {}
+    for metric in metrics:
+        by_bucket.setdefault(str(metric["bucket_end"]), []).append(
+            (str(metric["updated_at"]), float(metric["metric_value"]))
+        )
+    return {
+        bucket: [value for _, value in sorted(measurements)]
+        for bucket, measurements in by_bucket.items()
+    }
+
+
+def _restated_bucket(values: Dict[Any, List[float]]) -> List[float]:
+    """The one bucket whose measurements are not all identical."""
+    moved = [
+        measurements for measurements in values.values() if len(set(measurements)) > 1
+    ]
+    assert len(moved) == 1, f"expected exactly one bucket to move, got {moved}"
+    return moved[0]
+
+
+def test_metric_stability_detects_restated_settled_value(
+    test_id: str, dbt_project: DbtProject
+):
+    baseline = _rows()
+    args = {"change_since": ["last_check"]}
+
+    # The first run only establishes an initial measurement, so there is nothing
+    # to compare against yet.
+    assert _run(dbt_project, test_id, baseline, **args) == "pass"
+
+    # The second measures the same buckets again and the values are unchanged.
+    assert _run(dbt_project, test_id, baseline, **args) == "pass"
+
+    # Rewriting the value of an already-settled bucket is what the test exists to
+    # catch, even though the value itself is unremarkable next to other days.
+    restated = _rows({SETTLED_DAYS_AGO: BASE_AMOUNT * 2})
+    assert _run(dbt_project, test_id, restated, **args) == "fail"
+
+    # The bucket the test flagged must be the one that actually moved, and by
+    # the amount restated, so a wrong baseline cannot pass unnoticed.
+    measurements = _restated_bucket(_bucket_values(dbt_project, test_id))
+    assert measurements[0] == BASE_AMOUNT
+    assert measurements[-1] == BASE_AMOUNT * 2
+
+
+def test_metric_stability_first_check_catches_gradual_drift(
+    test_id: str, dbt_project: DbtProject
+):
+    """Drift too small to trip the threshold on any single step, but not overall.
+
+    This is the case that justifies having 'first_check' at all: comparing only
+    against the previous measurement never sees it.
+    """
+    args = {"change_since": ["first_check"], "max_change_percent": 15}
+    assert _run(dbt_project, test_id, _rows(), **args) == "pass"
+
+    # +10% against the original value: under the threshold either way.
+    assert _run(dbt_project, test_id, _rows({SETTLED_DAYS_AGO: 110}), **args) == "pass"
+
+    # A further +9% step, still under the threshold on its own, but now 20% away
+    # from where the bucket started.
+    assert _run(dbt_project, test_id, _rows({SETTLED_DAYS_AGO: 120}), **args) == "fail"
+
+    # Each step is under the threshold; only the distance from the first
+    # measurement crosses it.
+    measurements = _restated_bucket(_bucket_values(dbt_project, test_id))
+    assert measurements[0] == BASE_AMOUNT
+    assert measurements[-1] == 120
+    steps = [
+        later - earlier for earlier, later in pairwise(measurements) if later != earlier
+    ]
+    assert all(step / BASE_AMOUNT * 100 < 15 for step in steps), steps
+
+
+def test_metric_stability_last_check_ignores_gradual_drift(
+    test_id: str, dbt_project: DbtProject
+):
+    """The same drift, compared only against the previous run, stays invisible."""
+    args = {"change_since": ["last_check"], "max_change_percent": 15}
+    assert _run(dbt_project, test_id, _rows(), **args) == "pass"
+    assert _run(dbt_project, test_id, _rows({SETTLED_DAYS_AGO: 110}), **args) == "pass"
+    assert _run(dbt_project, test_id, _rows({SETTLED_DAYS_AGO: 120}), **args) == "pass"
+
+
+def test_metric_stability_records_history_across_runs(
+    test_id: str, dbt_project: DbtProject
+):
+    """Every run must add a measurement, not replace the previous one.
+
+    The comparison has nothing to compare unless earlier measurements survive in
+    data_monitoring_metrics. On adapters that roll back the test transaction an
+    INSERT-populated metrics table is discarded before the on-run-end flush, and
+    the test then passes forever without ever recording anything. Asserting on
+    the accumulating row counts catches that directly, where a pass/fail
+    assertion cannot tell "nothing changed" from "nothing was measured".
+    """
+    baseline = _rows()
+    counts = []
+    for _ in range(3):
+        assert _run(dbt_project, test_id, baseline) == "pass"
+        values = _bucket_values(dbt_project, test_id)
+        assert values, "no metrics were recorded at all"
+        counts.append(sum(len(m) for m in values.values()))
+
+    assert counts[0] > 0, counts
+    assert counts[1] > counts[0], counts
+    assert counts[2] > counts[1], counts
+
+    settled = [m for m in values.values() if len(m) == 3]
+    assert settled, f"no bucket was measured on all three runs: {values}"
+
+
+def test_metric_stability_detects_restatement_with_weekly_buckets(
+    test_id: str, dbt_project: DbtProject
+):
+    """A bucket longer than a day must still get a stable identity across runs.
+
+    The bucket grid is anchored on a value that moves by a day between runs, so
+    without snapping the anchor to the bucket period every measurement lands on
+    a fresh surrogate id, no bucket is ever measured twice and the test silently
+    never fires.
+    """
+    args = {
+        "time_bucket": {"period": "week", "count": 1},
+        "min_bucket_age": {"count": 1, "period": "week"},
+    }
+    restate_days_ago = 16
+
+    def weekly_rows(restated=None):
+        utc_today = datetime.utcnow().date()
+        rows = []
+        for days_ago in range(1, 36):
+            timestamp = datetime.combine(
+                utc_today - timedelta(days=days_ago), time(12, 0)
+            )
+            rows.append(
+                {
+                    TIMESTAMP_COLUMN: timestamp.strftime(DATE_FORMAT),
+                    VALUE_COLUMN: (
+                        restated
+                        if restated and days_ago == restate_days_ago
+                        else BASE_AMOUNT
+                    ),
+                    OTHER_VALUE_COLUMN: OTHER_BASE_AMOUNT,
+                }
+            )
+        return rows
+
+    assert _run(dbt_project, test_id, weekly_rows(), **args) == "pass"
+    assert _run(dbt_project, test_id, weekly_rows(), **args) == "pass"
+    assert (
+        _run(dbt_project, test_id, weekly_rows(restated=BASE_AMOUNT * 2), **args)
+        == "fail"
+    )
+
+
+def test_metric_stability_rejects_multi_step_buckets(
+    test_id: str, dbt_project: DbtProject
+):
+    """A count > 1 bucket cannot be given a stable identity, so it must raise."""
+    result = _run(
+        dbt_project,
+        test_id,
+        _rows(),
+        time_bucket={"period": "day", "count": 3},
+    )
+    assert result == "error"
+
+
+# Explicit ids: the value ends up in the seed relation name, and a "." in one
+# is rejected by the Hive metastore behind Trino.
+@pytest.mark.parametrize(
+    "days_back", [pytest.param(1, id="whole"), pytest.param(0.5, id="fractional")]
+)
+def test_metric_stability_rejects_too_short_window(
+    test_id: str, dbt_project: DbtProject, days_back
+):
+    """A window too short to outlast settling must raise, not pass vacuously.
+
+    The query truncates days_back to whole days, so a fractional value has to be
+    rejected on its effective size rather than on the number as written.
+    """
+    result = _run(
+        dbt_project,
+        test_id,
+        _rows(),
+        min_bucket_age={"count": 1, "period": "hour"},
+        days_back=days_back,
+    )
+    assert result == "error"
+
+
+def test_metric_stability_ignores_measurements_taken_while_settling(
+    test_id: str, dbt_project: DbtProject
+):
+    """A bucket's own settling must not become the 'first_check' baseline.
+
+    The first measurements of a bucket are taken while late records are still
+    arriving, which is the period min_bucket_age exists to exclude. If they are
+    used as the baseline, every later comparison carries that settling as a
+    permanent offset and real drift is buried under it.
+    """
+    args = {"change_since": ["first_check"], "max_change_percent": 15}
+    assert _run(dbt_project, test_id, _rows(), **args) == "pass"
+
+    # Backdate this run's measurements into the settling window and move their
+    # values far away. Were they still eligible as a baseline, the next run
+    # would compare 100 against 10 and report a 900% change.
+    if dbt_project.target == "clickhouse":
+        # ClickHouse only supports updates as (asynchronous) mutations.
+        update_clause = "ALTER TABLE {{ ref('data_monitoring_metrics') }} UPDATE"
+        update_suffix = "SETTINGS mutations_sync = 1"
+    else:
+        update_clause = "UPDATE {{ ref('data_monitoring_metrics') }} SET"
+        update_suffix = ""
+    dbt_project.run_query(
+        f"""
+        {update_clause} metric_value = 10, updated_at = bucket_end
+        WHERE full_table_name LIKE '%{test_id.upper()}'
+        AND metric_name = 'sum'
+        {update_suffix}
+        """
+    )
+
+    assert _run(dbt_project, test_id, _rows(), **args) == "pass"
+
+
+def test_metric_stability_ignores_unsettled_buckets(
+    test_id: str, dbt_project: DbtProject
+):
+    baseline = _rows()
+    assert _run(dbt_project, test_id, baseline) == "pass"
+    assert _run(dbt_project, test_id, baseline) == "pass"
+
+    # Recent data is expected to keep moving as late records arrive, so a change
+    # inside min_bucket_age must not be reported.
+    restated = _rows({UNSETTLED_DAYS_AGO: BASE_AMOUNT * 2})
+    assert _run(dbt_project, test_id, restated) == "pass"
+
+
+def test_metric_stability_tolerates_change_within_threshold(
+    test_id: str, dbt_project: DbtProject
+):
+    args = {"max_change_percent": 25}
+    assert _run(dbt_project, test_id, _rows(), **args) == "pass"
+    assert _run(dbt_project, test_id, _rows(), **args) == "pass"
+
+    # A 10% restatement sits under the 25% tolerance and should be allowed, which
+    # is what makes one relative threshold usable across metrics of very
+    # different magnitudes.
+    within = _rows({SETTLED_DAYS_AGO: int(BASE_AMOUNT * 1.1)})
+    assert _run(dbt_project, test_id, within, **args) == "pass"
+
+    beyond = _rows({SETTLED_DAYS_AGO: BASE_AMOUNT * 2})
+    assert _run(dbt_project, test_id, beyond, **args) == "fail"
+
+
+def test_metric_stability_detects_restatement_in_any_column(
+    test_id: str, dbt_project: DbtProject
+):
+    """Every monitored column is compared against this run's own measurements.
+
+    Collecting per column into separate tables would leave all but the last
+    column comparing against the previous run, so a restatement in an earlier
+    column would surface a run late.
+    """
+    args = {"columns": [VALUE_COLUMN, OTHER_VALUE_COLUMN]}
+    baseline = _rows()
+    assert _run(dbt_project, test_id, baseline, **args) == "pass"
+    assert _run(dbt_project, test_id, baseline, **args) == "pass"
+
+    # Restate the first of the two columns, which is the one a per-column table
+    # would have left stale.
+    restated = _rows({SETTLED_DAYS_AGO: BASE_AMOUNT * 2})
+    assert _run(dbt_project, test_id, restated, **args) == "fail"
+
+    # The restated column moved, and the other one did not.
+    measurements = _restated_bucket(_bucket_values(dbt_project, test_id))
+    assert measurements[0] == BASE_AMOUNT
+    assert measurements[-1] == BASE_AMOUNT * 2
+    other = _bucket_values(dbt_project, test_id, OTHER_VALUE_COLUMN)
+    assert all(len(set(m)) == 1 for m in other.values()), other
+
+
+def _samples(dbt_project: DbtProject, test_id: str):
+    test_id = test_id.replace("[", "_").replace("]", "_")
+    return [
+        {key.lower(): value for key, value in json.loads(row["result_row"]).items()}
+        for row in dbt_project.run_query(dbt_project.samples_query(test_id))
+    ]
+
+
+@pytest.mark.parametrize("baseline", ["last_check", "first_check"])
+def test_metric_stability_quoted_column_details(
+    test_id: str, dbt_project: DbtProject, baseline: str
+):
+    args = {"columns": ['"amount"'], "change_since": [baseline]}
+    assert _run(dbt_project, test_id, _rows(), **args) == "pass"
+    assert _run(dbt_project, test_id, _rows({SETTLED_DAYS_AGO: 200}), **args) == "fail"
+    samples = _samples(dbt_project, test_id)
+    assert len(samples) == 1
+    sample = samples[0]
+    assert sample["change_type"] == "value_changed"
+    assert float(sample["metric_value"]) == 200
+    assert float(sample["previous_value"]) == 100
+    assert float(sample["initial_value"]) == 100
+    assert sample["bucket_end"]
+    assert sample["previous_measured_at"]
+    assert sample["initial_measured_at"]
+    # A repeated corrected value is accepted only by the moving baseline.
+    expected = "pass" if baseline == "last_check" else "fail"
+    assert (
+        _run(dbt_project, test_id, _rows({SETTLED_DAYS_AGO: 200}), **args) == expected
+    )
+
+
+@pytest.mark.parametrize("dimensions", [[], ["other_amount"]])
+def test_metric_stability_reports_disappearing_bucket(
+    test_id: str, dbt_project: DbtProject, dimensions
+):
+    assert _run(dbt_project, test_id, _rows(), dimensions=dimensions) == "pass"
+    rows = _rows()
+    del rows[SETTLED_DAYS_AGO - 1]
+    assert _run(dbt_project, test_id, rows, dimensions=dimensions) == "fail"
+    samples = _samples(dbt_project, test_id)
+    assert len(samples) == 1
+    assert samples[0]["change_type"] == "missing_bucket"
+    assert samples[0]["metric_value"] is None
+    assert float(samples[0]["previous_value"]) == BASE_AMOUNT
+    # Absence is not accepted as a new numeric baseline on the next run.
+    assert _run(dbt_project, test_id, rows, dimensions=dimensions) == "fail"
+    assert _run(dbt_project, test_id, _rows(), dimensions=dimensions) == "pass"
+
+
+def test_metric_stability_ignores_disappearance_off_window(
+    test_id: str, dbt_project: DbtProject
+):
+    assert _run(dbt_project, test_id, _rows(), days_back=10) == "pass"
+    # Keep only the recent rows; historical measurements still exist but the
+    # default window no longer covers the deleted older buckets.
+    assert _run(dbt_project, test_id, _rows()[:3]) == "pass"
+
+
+def test_metric_stability_reports_disappearing_dimension(
+    test_id: str, dbt_project: DbtProject
+):
+    rows = _rows()
+    rows.append({**rows[SETTLED_DAYS_AGO - 1], OTHER_VALUE_COLUMN: 999})
+    args = {"dimensions": [OTHER_VALUE_COLUMN]}
+    assert _run(dbt_project, test_id, rows, **args) == "pass"
+    assert _run(dbt_project, test_id, _rows(), **args) == "fail"
+    samples = _samples(dbt_project, test_id)
+    assert len(samples) == 1
+    assert samples[0]["change_type"] == "missing_bucket"
+    assert "999" in str(samples[0]["dimension_value"])
+
+
+def test_metric_stability_skips_unscanned_buckets(
+    test_id: str, dbt_project: DbtProject
+):
+    assert _run(dbt_project, test_id, _rows(), days_back=6) == "pass"
+    # Sources use the incremental backfill window. Older buckets have history
+    # and remain in days_back, but are not rescanned with backfill_days=3.
+    assert (
+        _run(dbt_project, test_id, _rows()[:3], days_back=6, backfill_days=3) == "pass"
+    )
diff --git a/macros/edr/data_monitoring/monitors_query/metric_stability_query.sql b/macros/edr/data_monitoring/monitors_query/metric_stability_query.sql
new file mode 100644
index 000000000..6b31c6049
--- /dev/null
+++ b/macros/edr/data_monitoring/monitors_query/metric_stability_query.sql
@@ -0,0 +1,218 @@
+{# Compare each settled bucket's current value with its previous/first retained
+   measurement. Missing current measurements fail only within the rescan window. #}
+{% macro metric_stability_query(
+    test_metrics_table_relations,
+    full_table_name,
+    metric_names,
+    metric_properties,
+    detection_end,
+    days_back,
+    min_bucket_age,
+    max_change_percent=0,
+    change_since=["last_check"],
+    column_names=none,
+    data_monitoring_metrics_table=none,
+    measurement_windows=none
+) %}
+    {%- if not data_monitoring_metrics_table %}
+        {%- set data_monitoring_metrics_table = elementary.get_elementary_relation(
+            "data_monitoring_metrics"
+        ) %}
+    {%- endif %}
+
+    {%- set bucket_period = metric_properties.time_bucket.period %}
+
+    {# Only compare settled buckets within days_back. #}
+    {%- set age_kwargs = {min_bucket_age.period ~ "s": min_bucket_age.count} %}
+    {%- set max_bucket_end = detection_end - modules.datetime.timedelta(**age_kwargs) %}
+    {%- set min_bucket_end = detection_end - modules.datetime.timedelta(
+        days=days_back | int
+    ) %}
+    {%- set max_bucket_end_expr = elementary.edr_date_trunc(
+        bucket_period,
+        elementary.edr_cast_as_timestamp(
+            elementary.edr_datetime_to_sql(max_bucket_end)
+        ),
+    ) %}
+    {%- set min_bucket_end_expr = elementary.edr_date_trunc(
+        bucket_period,
+        elementary.edr_cast_as_timestamp(
+            elementary.edr_datetime_to_sql(min_bucket_end)
+        ),
+    ) %}
+    {# Baselines must also have been measured after the bucket settled. #}
+    {%- set settled_at_expr = elementary.edr_cast_as_timestamp(
+        elementary.edr_timeadd(
+            min_bucket_age.period, min_bucket_age.count, "bucket_end"
+        )
+    ) %}
+    {% set history_window %}
+        bucket_end > {{ min_bucket_end_expr }}
+        and bucket_end <= {{ max_bucket_end_expr }}
+        and updated_at >= {{ settled_at_expr }}
+    {% endset %}
+
+    {# Suppress floating-point aggregation noise; zero baselines are handled separately. #}
+    {%- set change_percent_noise_floor = 0.000000001 %}
+    {%- set change_threshold = "%.10f" | format(
+        [max_change_percent, change_percent_noise_floor] | max
+    ) %}
+
+    {%- set exceeds_conditions = [] %}
+    {%- set baseline_columns = [] %}
+    {%- for baseline in change_since %}
+        {%- set baseline_column = baseline ~ "_value" %}
+        {%- if baseline_column not in baseline_columns %}
+            {%- do baseline_columns.append(baseline_column) %}
+            {% set exceeds_condition %}
+                ({{ baseline_column }} is not null and (
+                    ({{ baseline_column }} = 0 and measured_value != 0)
+                    or ({{ baseline_column }} != 0 and
+                        {{ elementary.metric_stability_change_percent(baseline_column) }} > {{ change_threshold }})
+                ))
+            {% endset %}
+            {% do exceeds_conditions.append(exceeds_condition) %}
+        {%- endif %}
+    {%- endfor %}
+
+    {%- set metric_stability_query %}
+        with metrics_history as (
+
+            select id, full_table_name, column_name, metric_name, metric_type,
+                   bucket_start, bucket_end, bucket_duration_hours,
+                   metric_value, updated_at, dimension, dimension_value,
+                   0 as is_current
+            from {{ data_monitoring_metrics_table }}
+            where
+                upper(full_table_name) = upper('{{ full_table_name }}')
+                {%- if column_names %}
+                    {#- metric_properties does not carry the column, so without
+                        this a test picks up history for every other column
+                        monitored on the same table with the same properties. -#}
+                    and upper(column_name) in {{ elementary.strings_list_to_tuple(column_names | map("upper") | list) }}
+                {%- endif %}
+                and metric_name in {{ elementary.strings_list_to_tuple(metric_names) }}
+                and metric_properties = {{ elementary.dict_to_quoted_json(metric_properties) }}
+                and {{ history_window }}
+                {# History outside a column's actual rescan cannot establish
+                   absence. Restrict it before selecting the newest version. #}
+                {%- if measurement_windows %}
+                    and (
+                    {%- for window in measurement_windows %}
+                        (upper(column_name) = upper({{ elementary.edr_quote(window.column_name) }})
+                         and bucket_start >= {{ window.min_bucket_start }}
+                         and bucket_end <= {{ window.max_bucket_end }})
+                        {% if not loop.last %} or {% endif %}
+                    {%- endfor %}
+                    )
+                {%- endif %}
+
+            {%- for test_metrics_table_relation in test_metrics_table_relations %}
+
+            union all
+
+            select id, full_table_name, column_name, metric_name, metric_type,
+                   bucket_start, bucket_end, bucket_duration_hours,
+                   metric_value, updated_at, dimension, dimension_value,
+                   1 as is_current
+            from {{ test_metrics_table_relation }}
+            where {{ history_window }}
+            {%- endfor %}
+
+        ),
+
+        versioned_metrics as (
+
+            {#- Each measurement carries the values it is compared against,
+                named after the `change_since` baseline it serves. None of these
+                names is reused as an output alias of the final select:
+                ClickHouse resolves a select alias anywhere in the same select
+                list, so an output `metric_value` shadows the source column for
+                its sibling expressions and reports NULL baselines. -#}
+            select
+                id, full_table_name, column_name, metric_name, metric_type,
+                bucket_start, bucket_end, bucket_duration_hours,
+                updated_at, dimension, dimension_value, is_current,
+                metric_value as measured_value,
+                {{ elementary.lag("metric_value") }} over (
+                    partition by id order by updated_at
+                ) as last_check_value,
+                {{ elementary.lag("updated_at") }} over (
+                    partition by id order by updated_at
+                ) as last_check_at,
+                first_value(metric_value) over (
+                    partition by id order by updated_at
+                    rows between unbounded preceding and current row
+                ) as first_check_value,
+                first_value(updated_at) over (
+                    partition by id order by updated_at
+                    rows between unbounded preceding and current row
+                ) as first_check_at,
+                row_number() over (
+                    partition by id order by updated_at desc
+                ) as recency
+            from metrics_history
+
+        ),
+
+        latest_measurement as (
+
+            {#- One row per bucket: its newest measurement, carrying the values
+                it is being compared against. -#}
+            select * from versioned_metrics where recency = 1
+
+        )
+
+        {#- This select is frozen into a table, so every timestamp it carries out
+            of the metric history is cast to the precision that table accepts.
+            Athena reads timestamp(6) from the history and writes millisecond
+            columns, and rejects the CTAS otherwise. -#}
+        select
+            id as metric_id,
+            full_table_name,
+            column_name,
+            metric_name,
+            metric_type,
+            {{ elementary.edr_cast_as_timestamp("bucket_start") }} as bucket_start,
+            {{ elementary.edr_cast_as_timestamp("bucket_end") }} as bucket_end,
+            bucket_duration_hours,
+            dimension,
+            dimension_value,
+            {{ elementary.edr_cast_as_timestamp(
+                "case when is_current = 0"
+                ~ " then " ~ elementary.edr_cast_as_timestamp(elementary.edr_quote(elementary.run_started_at_as_string()))
+                ~ " else updated_at end"
+            ) }} as measured_at,
+            case when is_current = 0 then 'missing_bucket' else 'value_changed' end as change_type,
+            case when is_current = 1 then measured_value end as metric_value,
+            case when is_current = 0 then measured_value else last_check_value end as previous_value,
+            {{ elementary.edr_cast_as_timestamp(
+                "case when is_current = 0 then updated_at else last_check_at end"
+            ) }} as previous_measured_at,
+            {{ elementary.edr_cast_as_timestamp("first_check_at") }}
+            as initial_measured_at,
+            first_check_value as initial_value,
+            case when is_current = 1 then measured_value - last_check_value end as change_since_last_check,
+            case when is_current = 1 then measured_value - first_check_value end as change_since_first_check,
+            case
+                when is_current = 1 and last_check_value is not null and last_check_value != 0
+                then {{ elementary.metric_stability_change_percent("last_check_value") }}
+            end as change_percent_since_last_check,
+            case
+                when is_current = 1 and first_check_value is not null and first_check_value != 0
+                then {{ elementary.metric_stability_change_percent("first_check_value") }}
+            end as change_percent_since_first_check
+        from latest_measurement
+        where is_current = 0 or {{ exceeds_conditions | join(" or ") }}
+    {%- endset %}
+    {%- do return(metric_stability_query) %}
+{% endmacro %}
+
+
+{#
+  Relative change from a baseline column, in percentage points. Shared by the
+  WHERE predicate and the reported columns so the two cannot drift apart.
+#}
+{% macro metric_stability_change_percent(baseline_column) -%}
+    abs(measured_value - {{ baseline_column }}) / abs({{ baseline_column }}) * 100.0
+{%- endmacro %}
diff --git a/macros/edr/tests/test_metric_stability.sql b/macros/edr/tests/test_metric_stability.sql
new file mode 100644
index 000000000..5eab81833
--- /dev/null
+++ b/macros/edr/tests/test_metric_stability.sql
@@ -0,0 +1,561 @@
+{# Detect restatements of settled bucket metrics within the observation window.
+   Configuration, baseline behavior, and limits:
+   https://docs.elementary-data.com/data-tests/metric-stability #}
+{% test metric_stability(
+    model,
+    columns,
+    metrics,
+    timestamp_column,
+    min_bucket_age,
+    change_since=["last_check"],
+    max_change_percent=0,
+    time_bucket=none,
+    where_expression=none,
+    days_back=none,
+    backfill_days=none,
+    dimensions=none
+) %}
+    {{ config(tags=["elementary-tests"]) }}
+
+    {% if not (
+        execute
+        and elementary.is_test_command()
+        and elementary.is_elementary_enabled()
+    ) %}
+        {% do return(elementary.no_results_query()) %}
+    {% endif %}
+
+    {% set arguments = elementary._parse_and_validate_metric_stability_arguments(
+        model,
+        columns,
+        metrics,
+        timestamp_column,
+        min_bucket_age,
+        change_since,
+        max_change_percent,
+        days_back,
+        backfill_days,
+    ) %}
+    {% set model_relation = arguments.model_relation %}
+    {% set model_graph_node = arguments.model_graph_node %}
+    {% set timestamp_column = arguments.timestamp_column %}
+    {% set metrics = arguments.metrics %}
+    {% set change_since = arguments.change_since %}
+    {%- if not dimensions %} {% set dimensions = [] %} {%- endif %}
+
+    {% set metric_properties = elementary.get_metric_properties(
+        model_graph_node,
+        timestamp_column,
+        where_expression,
+        time_bucket,
+        dimensions,
+        collected_by="metric_stability",
+    ) %}
+    {% set metric_names = metrics %}
+
+    {# Size the scan from the settling age and bucket duration. #}
+    {% set resolved_window = elementary.resolve_metric_stability_window(
+        model_graph_node,
+        min_bucket_age,
+        metric_properties.time_bucket,
+        days_back,
+        backfill_days,
+    ) %}
+    {% set days_back = resolved_window["days_back"] %}
+    {% set backfill_days = resolved_window["backfill_days"] %}
+
+    {% set test_table_name = elementary.get_elementary_test_table_name() %}
+    {% set (
+        database_name,
+        schema_name,
+    ) = elementary.get_package_database_and_schema("elementary") %}
+    {% set tests_schema_name = elementary.get_elementary_tests_schema(
+        database_name, schema_name
+    ) %}
+    {% set full_table_name = elementary.relation_to_full_name(model_relation) %}
+
+    {% set collected = elementary._collect_metric_stability_metrics(
+        model,
+        model_relation,
+        arguments.columns,
+        metric_properties,
+        days_back,
+        backfill_days,
+        dimensions,
+        database_name,
+        tests_schema_name,
+        test_table_name,
+    ) %}
+    {% set detection_end = elementary.get_detection_end(none) %}
+    {% set metric_stability_query = elementary.metric_stability_query(
+        test_metrics_table_relations=collected.relations,
+        full_table_name=full_table_name,
+        metric_names=metric_names,
+        metric_properties=metric_properties,
+        detection_end=detection_end,
+        days_back=days_back,
+        min_bucket_age=min_bucket_age,
+        max_change_percent=max_change_percent,
+        change_since=change_since,
+        column_names=collected.columns,
+        measurement_windows=collected.windows,
+    ) %}
+    {{ elementary.debug_log("metric_stability_query - \n" ~ metric_stability_query) }}
+
+    {# Freeze failures before dbt executes the test and samples its results.
+       Use the normal sampling path so sample limits and privacy settings
+       still apply. This relation is cleaned up with the metrics tables. #}
+    {% set result_relation = elementary.create_elementary_test_table(
+        database_name,
+        tests_schema_name,
+        test_table_name,
+        "stability_results",
+        metric_stability_query,
+    ) %}
+    select *
+    from {{ result_relation }}
+
+{% endtest %}
+
+{# Collect one persistent CTAS table per column and return relations, scan
+   windows, and resolved names. Tables are unioned when comparing history. #}
+{% macro _collect_metric_stability_metrics(
+    model,
+    model_relation,
+    column_definitions,
+    metric_properties,
+    days_back,
+    backfill_days,
+    dimensions,
+    database_name,
+    tests_schema_name,
+    test_table_name
+) %}
+    {% set temp_table_relations = [] %}
+    {% set measurement_windows = [] %}
+    {% set resolved_columns = [] %}
+
+    {%- for column_obj_and_monitors in column_definitions %}
+        {% set resolved_column = column_obj_and_monitors["column"].name %}
+        {% do resolved_columns.append(resolved_column) %}
+        {% set column_monitors = column_obj_and_monitors["monitors"] %}
+
+        {%- set (
+            raw_min_bucket_start,
+            max_bucket_end,
+        ) = elementary.get_metric_buckets_min_and_max(
+            model_relation=model_relation,
+            backfill_days=backfill_days,
+            days_back=days_back,
+            metric_names=column_monitors,
+            column_name=resolved_column,
+            metric_properties=metric_properties,
+        ) %}
+        {# Keep bucket identities stable across runs, including weekly buckets. #}
+        {%- set min_bucket_start = elementary.edr_date_trunc(
+            metric_properties.time_bucket.period,
+            elementary.edr_cast_as_timestamp(raw_min_bucket_start),
+        ) %}
+        {%- do measurement_windows.append(
+            {
+                "column_name": resolved_column,
+                "min_bucket_start": min_bucket_start,
+                "max_bucket_end": elementary.edr_cast_as_timestamp(
+                    max_bucket_end
+                ),
+            }
+        ) %}
+        {%- set this_column_metrics = [] %}
+        {%- for monitor in column_monitors %}
+            {%- do this_column_metrics.append({"name": monitor, "type": monitor}) %}
+        {%- endfor %}
+        {%- set column_monitoring_query = elementary.column_monitoring_query(
+            model,
+            model_relation,
+            min_bucket_start,
+            max_bucket_end,
+            days_back,
+            column_obj_and_monitors["column"],
+            this_column_metrics,
+            metric_properties,
+            dimensions,
+        ) %}
+        {%- do temp_table_relations.append(
+            elementary.create_elementary_test_table(
+                database_name,
+                tests_schema_name,
+                test_table_name,
+                "metrics_" ~ loop.index0,
+                column_monitoring_query,
+            )
+        ) %}
+    {%- endfor %}
+
+    {# Register every column table for history persistence at on-run-end. #}
+    {% set metrics_tables_cache = (
+        elementary.get_cache("tables").get("metrics").get("relations")
+    ) %}
+    {%- for temp_table_relation in temp_table_relations %}
+        {% do metrics_tables_cache.append(temp_table_relation) %}
+    {%- endfor %}
+
+    {% do return(
+        {
+            "relations": temp_table_relations,
+            "windows": measurement_windows,
+            "columns": resolved_columns,
+        }
+    ) %}
+{% endmacro %}
+
+{% macro validate_min_bucket_age(min_bucket_age) %}
+    {%- set valid_periods = ["day", "week", "hour", "minute", "second"] %}
+    {%- if not min_bucket_age or min_bucket_age is not mapping %}
+        {% do exceptions.raise_compiler_error(
+            "min_bucket_age is required and must be a mapping. Expected format:   min_bucket_age:     count: int     period: string"
+        ) %}
+    {%- endif %}
+    {%- for key in min_bucket_age %}
+        {%- if key not in ["count", "period"] %}
+            {% do exceptions.raise_compiler_error(
+                "Found invalid key in min_bucket_age: '"
+                ~ key
+                ~ "'. Supported keys: count, period."
+            ) %}
+        {%- endif %}
+    {%- endfor %}
+    {%- if min_bucket_age.period not in valid_periods %}
+        {% do exceptions.raise_compiler_error(
+            "Unsupported min_bucket_age period '"
+            ~ min_bucket_age.period
+            ~ "'. Supported periods: "
+            ~ valid_periods
+            | join(", ")
+            ~ ". time_bucket also accepts month, quarter and year; express an age over those in days, e.g. {count: 60, period: day}."
+        ) %}
+    {%- endif %}
+    {%- if min_bucket_age.count is not integer or min_bucket_age.count < 1 %}
+        {% do exceptions.raise_compiler_error(
+            "min_bucket_age count must be a positive integer, got '"
+            ~ min_bucket_age.count
+            ~ "'."
+        ) %}
+    {%- endif %}
+{% endmacro %}
+
+
+{# Return days_back/backfill_days sized for settled buckets; reject windows
+   below the required minimum. Backfill applies to sources/incremental models. #}
+{% macro resolve_metric_stability_window(
+    model_graph_node, min_bucket_age, time_bucket, days_back, backfill_days
+) %}
+    {%- set age_kwargs = {min_bucket_age.period ~ "s": min_bucket_age.count} %}
+    {%- set age_days = (
+        modules.datetime.timedelta(**age_kwargs).total_seconds() / 86400.0
+    ) %}
+
+    {%- if time_bucket.count | int != 1 %}
+        {% do exceptions.raise_compiler_error(
+            "metric_stability requires a time_bucket count of 1, got "
+            ~ time_bucket.count
+            ~ ". A multi-step bucket cannot be measured on a stable grid across runs, so the test would never report a change."
+        ) %}
+    {%- endif %}
+
+    {# Calendar periods use nominal day lengths for sizing only. #}
+    {%- set period_days = {
+        "second": 1.0 / 86400.0,
+        "minute": 1.0 / 1440.0,
+        "hour": 1.0 / 24.0,
+        "day": 1.0,
+        "week": 7.0,
+        "month": 30.0,
+        "quarter": 91.0,
+        "year": 365.0,
+    } %}
+    {%- set bucket_days = period_days.get(time_bucket.period | lower) %}
+    {%- if not bucket_days %}
+        {% do exceptions.raise_compiler_error(
+            "Unsupported time_bucket period for metric_stability: '"
+            ~ time_bucket.period
+            ~ "'."
+        ) %}
+    {%- endif %}
+
+    {# The window has to outlast settling, or nothing is ever both settled and
+       still measured. Take the widest of three floors:
+         - twice the age, so a bucket is observed over a stretch of runs rather
+           than measured once and never compared;
+         - two whole buckets past the age, or the settled band is narrower than
+           a single bucket;
+         - one whole day past the age. days_back is counted in whole days and
+           the metrics scan starts at midnight, so this keeps a sub-day age from
+           collapsing the window to a sliver of a day, and it is what makes the
+           floor at least 2 for every age, so a days_back the query would
+           truncate to 0 or 1 days is rejected below instead of silently
+           producing an empty comparison window. #}
+    {%- set twice_the_age = (age_days * 2) | round(0, "ceil") | int %}
+    {%- set two_buckets_past_age = (
+        (age_days + 2 * bucket_days) | round(0, "ceil") | int
+    ) %}
+    {%- set one_day_past_age = (age_days + 1) | round(0, "ceil") | int %}
+    {%- set derived = [twice_the_age, two_buckets_past_age, one_day_past_age] | max %}
+    {%- set age_description = (
+        min_bucket_age.count
+        ~ " "
+        ~ min_bucket_age.period
+        ~ ("s" if min_bucket_age.count > 1 else "")
+    ) %}
+
+    {%- set uses_backfill_window = elementary.is_incremental_model(
+        model_graph_node, source_included=true
+    ) and not elementary.get_config_var("force_metrics_backfill") %}
+
+    {%- if days_back is none %} {%- set resolved_days_back = derived %}
+    {%- else %}
+        {%- set resolved_days_back = days_back %}
+        {%- if resolved_days_back < derived %}
+            {% do exceptions.raise_compiler_error(
+                "days_back is "
+                ~ resolved_days_back
+                ~ ", which does not leave room for whole buckets past a min_bucket_age of "
+                ~ age_description
+                ~ ", so no bucket is ever both settled and still measured and the test can never report a change. Use at least "
+                ~ derived
+                ~ ", or omit days_back to have it derived."
+            ) %}
+        {%- endif %}
+    {%- endif %}
+
+    {%- if backfill_days is none %}
+        {%- set resolved_backfill_days = resolved_days_back %}
+    {%- else %}
+        {%- set resolved_backfill_days = backfill_days %}
+        {%- if uses_backfill_window and resolved_backfill_days < derived %}
+            {% do exceptions.raise_compiler_error(
+                "backfill_days is "
+                ~ resolved_backfill_days
+                ~ ", which does not extend past a min_bucket_age of "
+                ~ age_description
+                ~ ". On incremental models and sources backfill_days sets how far back buckets are re-measured, so those buckets freeze before they become eligible to check. Use at least "
+                ~ derived
+                ~ ", or omit backfill_days to have it derived."
+            ) %}
+        {%- endif %}
+    {%- endif %}
+
+    {%- do return(
+        {
+            "days_back": resolved_days_back,
+            "backfill_days": resolved_backfill_days,
+        }
+    ) %}
+{% endmacro %}
+
+
+{# Validate and normalize arguments; return resolved model, timestamp, and columns. #}
+{% macro _parse_and_validate_metric_stability_arguments(
+    model,
+    columns,
+    metrics,
+    timestamp_column,
+    min_bucket_age,
+    change_since,
+    max_change_percent,
+    days_back,
+    backfill_days
+) %}
+    {%- if columns is string %} {% set columns = [columns] %} {%- endif %}
+    {%- if metrics is string %} {% set metrics = [metrics] %} {%- endif %}
+    {%- if change_since is string %}
+        {% set change_since = [change_since] %}
+    {%- endif %}
+    {%- if columns %}
+        {%- set seen_columns = [] %}
+        {%- set deduped_columns = [] %}
+        {%- for column_name in columns %}
+            {%- set key = column_name | trim('"') | lower %}
+            {%- if key not in seen_columns %}
+                {%- do seen_columns.append(key) %}
+                {%- do deduped_columns.append(column_name) %}
+            {%- endif %}
+        {%- endfor %}
+        {%- set columns = deduped_columns %}
+    {%- endif %}
+
+    {%- if not change_since %}
+        {{
+            exceptions.raise_compiler_error(
+                "metric_stability requires at least one baseline in `change_since`: 'last_check', 'first_check', or both."
+            )
+        }}
+    {%- endif %}
+
+    {%- if max_change_percent is not number %}
+        {{
+            exceptions.raise_compiler_error(
+                "max_change_percent must be a number, got '"
+                ~ max_change_percent
+                ~ "'. Write it unquoted, e.g. max_change_percent: 25."
+            )
+        }}
+    {%- endif %}
+    {%- if max_change_percent < 0 %}
+        {{
+            exceptions.raise_compiler_error(
+                "max_change_percent must be non-negative."
+            )
+        }}
+    {%- endif %}
+    {%- for arg_name, arg_value in [
+        ("days_back", days_back),
+        ("backfill_days", backfill_days),
+    ] %}
+        {%- if arg_value is not none and arg_value is not number %}
+            {{
+                exceptions.raise_compiler_error(
+                    arg_name
+                    ~ " must be a number, got '"
+                    ~ arg_value
+                    ~ "'. Write it unquoted, e.g. "
+                    ~ arg_name
+                    ~ ": 30."
+                )
+            }}
+        {%- endif %}
+    {%- endfor %}
+
+    {%- if not columns %}
+        {{
+            exceptions.raise_compiler_error(
+                "metric_stability requires at least one column in `columns`."
+            )
+        }}
+    {%- endif %}
+
+    {%- if not metrics %}
+        {{
+            exceptions.raise_compiler_error(
+                "metric_stability requires at least one metric type in `metrics`."
+            )
+        }}
+    {%- endif %}
+
+    {%- set available_column_monitors = elementary.get_available_column_monitors() %}
+    {%- for metric_type in metrics %}
+        {%- if metric_type not in available_column_monitors %}
+            {{
+                exceptions.raise_compiler_error(
+                    "Unsupported column metric: '"
+                    ~ metric_type
+                    ~ "'. Supported metrics are: "
+                    ~ available_column_monitors
+                    | join(", ") ~ "."
+                )
+            }}
+        {%- endif %}
+    {%- endfor %}
+
+    {%- for baseline in change_since %}
+        {%- if baseline not in ["last_check", "first_check"] %}
+            {{
+                exceptions.raise_compiler_error(
+                    "Unsupported `change_since` value '"
+                    ~ baseline
+                    ~ "'. Supported values are 'last_check' and 'first_check'."
+                )
+            }}
+        {%- endif %}
+    {%- endfor %}
+
+    {% do elementary.validate_min_bucket_age(min_bucket_age) %}
+
+    {% set model_relation = elementary.get_model_relation_for_test(
+        model, elementary.get_test_model()
+    ) %}
+    {%- if not model_relation %}
+        {{ exceptions.raise_compiler_error("Unsupported model: " ~ model) }}
+    {%- endif %}
+
+    {%- if elementary.is_ephemeral_model(model_relation) %}
+        {{
+            exceptions.raise_compiler_error(
+                "Test not supported for ephemeral models: "
+                ~ model_relation.identifier
+            )
+        }}
+    {%- endif %}
+
+    {% set model_graph_node = elementary.get_model_graph_node(model_relation) %}
+    {% set timestamp_column = elementary.get_test_argument(
+        "timestamp_column", timestamp_column, model_graph_node
+    ) %}
+    {%- if not timestamp_column %}
+        {{
+            exceptions.raise_compiler_error(
+                "metric_stability requires a `timestamp_column`, either on the test or in the model's elementary config."
+            )
+        }}
+    {%- endif %}
+
+    {% set timestamp_column_data_type = (
+        elementary.find_normalized_data_type_for_column(
+            model_relation, timestamp_column
+        )
+    ) %}
+    {%- if not elementary.is_column_timestamp(
+        model_relation, timestamp_column, timestamp_column_data_type
+    ) %}
+        {{
+            exceptions.raise_compiler_error(
+                "Column '"
+                ~ timestamp_column
+                ~ "' is not a timestamp type. metric_stability buckets data over time and requires a timestamp column."
+            )
+        }}
+    {%- endif %}
+
+    {% set full_table_name = elementary.relation_to_full_name(model_relation) %}
+    {% set column_definitions = [] %}
+    {% for column_name in columns %}
+        {%- set column_obj_and_monitors = elementary.get_column_obj_and_monitors(
+            model_relation, column_name, metrics
+        ) -%}
+        {%- if not column_obj_and_monitors %}
+            {{
+                exceptions.raise_compiler_error(
+                    "Unable to find column `"
+                    ~ column_name
+                    ~ "` in `"
+                    ~ full_table_name
+                    ~ "`."
+                )
+            }}
+        {%- endif %}
+        {%- set resolved_column = column_obj_and_monitors["column"].name %}
+        {%- set column_monitors = column_obj_and_monitors["monitors"] %}
+        {%- if not column_monitors %}
+            {{
+                exceptions.raise_compiler_error(
+                    "None of the metrics " ~ metrics
+                    | join(", ")
+                    ~ " apply to column `"
+                    ~ column_name
+                    ~ "` given its data type."
+                )
+            }}
+        {%- endif %}
+
+        {% do column_definitions.append(column_obj_and_monitors) %}
+    {% endfor %}
+    {% do return(
+        {
+            "columns": column_definitions,
+            "metrics": metrics,
+            "change_since": change_since,
+            "model_relation": model_relation,
+            "model_graph_node": model_graph_node,
+            "timestamp_column": timestamp_column,
+        }
+    ) %}
+{% endmacro %}