feat(databricks): data quality monitoring from a contract's freshness and completeness SLOs

CHANGELOG.md

@@ -50,6 +50,26 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   column the dbt model leaves nullable, say) is reported by the next `fluid apply`, and the
   script needs a SQL warehouse that runs SQL scripting. A dbt build that
   `--mode amend-and-build` runs locally is followed by the same re-declaration.
+- **Databricks data quality monitoring from a contract's freshness and completeness SLOs.** An
+  expose bound to Databricks that states a freshness or completeness SLO (`qos.freshnessSLO`,
+  `qos.completenessTarget`, an `observability.defaultSLIs` freshness or completeness block, or a
+  `contract.dq.rules` entry of type `freshness`, `completeness` or `anomaly_detection`) now turns on
+  Databricks data quality monitoring (formerly anomaly detection) for the schema the product owns:
+  one `databricks_data_quality_monitor` per schema, created after its tables, with the product's
+  other tables there excluded by name. Catalog Explorer then shows a Quality indicator on the table,
+  and every scan lands in `system.data_quality_monitoring.table_results`. No new contract field:
+  `contract.dq.monitoring.enabled: false` opts an expose out. Databricks learns each table's bounds
+  from its history and takes no threshold, so the SLO values choose the tables and are not the
+  bounds it checks; `fluid apply` says so, and `fluid validate` warns about a `dq.monitoring.window`
+  other than one day. A monitor turned on outside forge-cli is adopted, not created over. If
+  Databricks refuses the monitor (no serverless compute, or no MANAGE on the schema), the apply
+  fails after the tables are created and names the opt-out first. Removing one (opting out, dropping
+  the SLOs, removing the expose) deletes its hidden scan job and what it has learned, never table
+  data, so it is not gated, and is printed. **Cost:** scans run on serverless compute billed as
+  `DATA_QUALITY_MONITORING`, once when monitoring is turned on and then as often as each table is
+  updated. Databricks Free Edition runs it (proven 2026-11-12), and its daily compute quota covers
+  these scans too: a workspace over the quota stops all compute for the rest of the day, so on Free
+  Edition opt out the products you do not need watched.
 
 ### Fixed
 

fluid_build/cli/_apply_opentofu_engine.py

@@ -270,8 +270,9 @@ def _apply_via_opentofu(args, logger: logging.Logger, contract: Optional[Mapping
         retired = _retired_definitions(changes, removals)
         if retired:
             cprint(
-                f"  {len(retired)} job(s), workspace file(s) or folder(s) removed (no table data "
-                "lost; not gated; a job's run history goes with it): " + ", ".join(retired)
+                f"  {len(retired)} job(s), workspace file(s), folder(s) or data quality "
+                "monitor(s) removed (no table data lost; not gated; a job's run history and "
+                "what a monitor has learned go with them): " + ", ".join(retired)
             )
         _report_plan_notices(plugin, contract, logger)
         # In-place updates that delete data — a dropped column — which the destroy
@@ -316,7 +317,7 @@ def _apply_via_opentofu(args, logger: logging.Logger, contract: Optional[Mapping
                 "opentofu apply raw output tail: %s",
                 _tail(apply_result.stderr or apply_result.stdout),
             )
-            raise _apply_failure(apply_result, str(workdir))
+            raise _apply_failure(apply_result, str(workdir), plugin)
         applied = runner.change_summary(apply_result)
         record(applied)
 
@@ -947,7 +948,10 @@ def _adopt_existing(
     for block in candidates:
         if block.to in in_state:
             continue
-        result = runner.tofu_import(workdir, block.to, block.id, env=env)
+        import_id = _import_id(block, workdir, env)
+        if import_id is None:
+            continue
+        result = runner.tofu_import(workdir, block.to, import_id, env=env)
         if result.ok:
             adopted += 1
             if held is not None:
@@ -965,6 +969,21 @@ def _adopt_existing(
         cprint(f"  brownfield:  adopted {adopted} pre-existing resource(s) into state")
 
 
+def _import_id(block: Any, workdir: str, env: Mapping[str, str]) -> Optional[str]:
+    """``block.id``, or for a block whose id the cloud assigns (``ImportBlock.id_from``) that id
+    completed from state, which the imports before it have just filled. ``None`` when the
+    resource it derives from is not in state: then neither exists yet, and there is nothing
+    to adopt."""
+    if getattr(block, "id_from", None) is None:
+        return block.id
+    address, attribute = block.id_from
+    for resource in runner.tofu_state_resources(workdir, env=env):
+        if resource.get("address") == address:
+            value = (resource.get("values") or {}).get(attribute)
+            return block.id.replace("{}", value) if isinstance(value, str) and value else None
+    return None
+
+
 @dataclass
 class _Imports:
     """What one run's brownfield adoption imported (:func:`_adopt_existing`)."""
@@ -1503,9 +1522,16 @@ def _report_suppressed_drift(
 #: no table data: a Databricks job (its run history goes with it) and a workspace
 #: file of a dbt project uploaded from the build's repository. Removing a build
 #: from the contract removes them, ungated, with a line saying so (founder
-#: decision for the Databricks provider, 2026-10-29).
+#: decision for the Databricks provider, 2026-10-29). A schema's data quality
+#: monitor is one too: deleting it deletes its hidden scan job and what it has
+#: learned, never a row (``iac/providers/databricks_quality.py``).
 RUN_DEFINITION_RESOURCE_TYPES = frozenset(
-    {"databricks_job", "databricks_workspace_file", "databricks_directory"}
+    {
+        "databricks_job",
+        "databricks_workspace_file",
+        "databricks_directory",
+        "databricks_data_quality_monitor",
+    }
 )
 
 
@@ -1635,19 +1661,22 @@ def _mentions_state_lock(text: str) -> bool:
     return "state lock" in low or "locking state" in low
 
 
-def _apply_failure(result: Any, workdir: str) -> CLIError:
+def _apply_failure(result: Any, workdir: str, plugin: Any = None) -> CLIError:
     """The error for a failed ``tofu apply``: what failed, on which resource, and the way on.
 
     ``-json`` puts every diagnostic on stdout as one JSON line, so the raw tail
     was a run of ``{"@level":...}`` objects. The diagnostics are read back into
     a summary, the resource address and the provider's detail; output that holds
     none (a crash, a timeout) falls back to the first ``Error:``, else the exit
     code. Tofu records state as it goes, so what was created before the failure
-    is normally tracked and a re-run converges: the suggestions say so.
+    is normally tracked and a re-run converges: the suggestions say so. A plugin's
+    optional ``apply_failure_hints(address, detail)`` puts its own way on first, for the
+    resource that failed.
     """
     diagnostics = _diagnostics(getattr(result, "events", None))
     context: Dict[str, Any] = {}
     locked = False
+    address = detail = ""
     if diagnostics:
         first = diagnostics[0]
         # Redacted before it is cut: a credential split at the limit (a URL that lost
@@ -1673,6 +1702,9 @@ def _apply_failure(result: Any, workdir: str) -> CLIError:
         locked = _mentions_state_lock(output)
     err = CLIError(1, "opentofu_apply_failed", context)
     hints = [f"To see what exists, run 'tofu state list' in {shlex.quote(workdir)}"]
+    plugin_hints = getattr(plugin, "apply_failure_hints", None)
+    if address and callable(plugin_hints):
+        hints = list(plugin_hints(address, detail)) + hints
     if locked:
         hints.append("The state is locked: wait for the other run, or 'tofu force-unlock <id>'")
     err.suggestions = list(err.suggestions) + hints

fluid_build/iac/base.py

@@ -66,6 +66,9 @@ class IacProviderPlugin(Protocol):
     * ``explain_plan_failure(output, contract, state)``: an
       :class:`UnsupportedBindingError` naming the fix for a failed ``tofu``
       run, or ``None``.
+    * ``apply_failure_hints(address, detail)``: suggestions, first in line, for
+      a ``tofu apply`` that failed on the resource at ``address``; ``[]`` when
+      the plugin has none.
     * ``recorded_regions(state)``: the regions the state's resources live in,
       so an apply pinned to another region is refused, not re-created there.
     * ``suppressed_drift(contract, prior_state)``: drift the module hides with

fluid_build/iac/importer.py

@@ -17,7 +17,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
-from typing import Any, Dict, List
+from typing import Any, Dict, List, Optional, Tuple
 
 
 @dataclass(frozen=True)
@@ -36,6 +36,12 @@ class ImportBlock:
     #: adopted (a Databricks grant is authoritative for its principal), so a
     #: failed import costs an honest plan, never a silent replace.
     best_effort: bool = False
+    #: ``(address, attribute)`` of the resource whose value in state completes ``id``
+    #: (its ``{}``): an import id the cloud assigns, such as a Databricks schema's UUID,
+    #: known only once that resource is in state. The import is skipped while it is not,
+    #: and the block is never rendered into a module (:func:`import_section`): an import
+    #: id there must be known before the plan.
+    id_from: Optional[Tuple[str, str]] = None
 
 
 def import_section(blocks: List[ImportBlock]) -> Dict[str, Any]:
@@ -44,6 +50,7 @@ def import_section(blocks: List[ImportBlock]) -> Dict[str, Any]:
     Returns an empty dict when there is nothing to import (so callers can
     unconditionally merge the result).
     """
-    if not blocks:
+    literal = [block for block in blocks if block.id_from is None]
+    if not literal:
         return {}
-    return {"import": [{"to": block.to, "id": block.id} for block in blocks]}
+    return {"import": [{"to": block.to, "id": block.id} for block in literal]}

fluid_build/iac/providers/databricks.py

@@ -87,6 +87,16 @@
   A predicate that is not one boolean expression over the table's columns is
   ``databricks-row-filter-invalid``; ``access: allow``, a principal with no SQL test and a text
   mask on a non-string column are refused too. ``fluid verify`` checks the bindings.
+* An expose that states a freshness or completeness SLO (``qos.freshnessSLO``,
+  ``qos.completenessTarget``, ``observability.defaultSLIs``, a ``contract.dq.rules``
+  entry of type ``freshness``, ``completeness`` or ``anomaly_detection``) turns on
+  Databricks data quality monitoring for its schema: one
+  ``databricks_data_quality_monitor`` per owned schema, the module's other tables there
+  excluded, ``contract.dq.monitoring.enabled: false`` to opt out
+  (``databricks_quality.py``). Databricks learns each table's bounds itself: the SLO
+  values pick the tables and are not thresholds the monitor checks. Its results are the
+  Quality indicator on the table in Catalog Explorer and
+  ``system.data_quality_monitoring.table_results``; removing it is not gated.
 * A dbt build that runs on Databricks (``execution.runtime.platform:
   databricks``) becomes a serverless ``databricks_job`` with a ``dbt_task``, and
   a local project its ``databricks_workspace_file`` uploads
@@ -211,6 +221,14 @@
     reconcile,
 )
 from .databricks_keys import lookup as lookup_key_tables
+from .databricks_quality import RESOURCE_TYPE as QUALITY_MONITOR
+from .databricks_quality import (
+    SchemaMonitor,
+    failure_hints,
+    monitor_notices,
+    plan_monitors,
+    window_warnings,
+)
 from .databricks_retention import Retention, check_expire
 
 _LOG = logging.getLogger(__name__)
@@ -355,6 +373,8 @@ class _Compiled:
     warnings: List[str] = field(default_factory=list)
     #: ``(entity type, full name, tag key, tag value)`` per tag assignment emitted.
     tags: List[Tuple[str, str, str, str]] = field(default_factory=list)
+    #: The data quality monitor emitted per schema (``databricks_quality.py``).
+    monitors: List[SchemaMonitor] = field(default_factory=list)
 
 
 #: Unity Catalog allows 50 tags on a securable ("Apply tags to Unity Catalog securable
@@ -1388,6 +1408,7 @@ def _compile(
         _emit_grants(compiled, table, cid, privileges.get(table.full_name, {}))
         _emit_tags(compiled, table, cid, contract)
     _check_own_keys(compiled, contract)
+    _emit_quality_monitors(compiled, cid)
     _merge_jobs(
         compiled,
         compile_dbt_jobs(
@@ -1607,15 +1628,60 @@ def _refuse_duplicate_tables(
         bound.setdefault(table.full_name, table.where)
 
 
+def _schema_key(cid: str, catalog: str, schema: str) -> str:
+    """The resource key of an owned schema, and of its data quality monitor."""
+    return safe_ident(f"{cid}_{catalog}_{schema}")
+
+
 def _keys(cid: str, table: _Table) -> Tuple[str, str, str]:
     """``(catalog, schema, table)`` resource keys of one table."""
     return (
         safe_ident(f"{cid}_{table.catalog}"),
-        safe_ident(f"{cid}_{table.catalog}_{table.schema}"),
+        _schema_key(cid, table.catalog, table.schema),
         safe_ident(f"{cid}_{table.catalog}_{table.schema}_{table.name}"),
     )
 
 
+def _emit_quality_monitors(compiled: _Compiled, cid: str) -> None:
+    """Databricks data quality monitoring on each owned schema one of its tables asks it for
+    (``databricks_quality.py``), keyed like the schema it watches.
+
+    A monitor someone turned on outside forge-cli is adopted, not created over: a second
+    create fails (``RESOURCE_ALREADY_EXISTS``, measured on Free Edition 2026-11-12). Its
+    import id is ``schema,<schema_id>``, a UUID Databricks assigns, so the import takes it
+    from the schema's state after the schema's own import (``ImportBlock.id_from``).
+
+    It is created after the schema's tables (``depends_on``): enabling it starts a scan, and
+    a table that scan misses may wait up to two weeks for its health indicator (Databricks
+    docs, "intelligent scanning").
+    """
+    compiled.warnings.extend(window_warnings(compiled.tables))
+    for monitor in plan_monitors(compiled.tables):
+        key = _schema_key(cid, monitor.catalog, monitor.schema)
+        body = monitor.body(tofu_ref(f"databricks_schema.{key}.schema_id"))
+        body["depends_on"] = [
+            f"databricks_sql_table.{_keys(cid, table)[2]}"
+            for table in compiled.tables
+            if (table.catalog, table.schema) == (monitor.catalog, monitor.schema)
+        ]
+        _claim(
+            compiled,
+            QUALITY_MONITOR,
+            key,
+            f"data quality monitor on schema {monitor.full_name}",
+            body,
+        )
+        compiled.imports.append(
+            ImportBlock(
+                f"{QUALITY_MONITOR}.{key}",
+                "schema,{}",
+                best_effort=True,
+                id_from=(f"databricks_schema.{key}", "schema_id"),
+            )
+        )
+        compiled.monitors.append(monitor)
+
+
 def _emit_table(
     compiled: _Compiled, table: _Table, cid: str, warehouse: str, region: str, *, dbt_owned: bool
 ) -> None:
@@ -2450,7 +2516,7 @@ def plan_notices(self, contract: Mapping[str, Any]) -> List[str]:
         """What the apply cannot do itself and someone must: ``USE CATALOG``; and the dbt jobs
         it creates paused until the tables they write are protected."""
         compiled = _compile(contract, os.environ)
-        notices = _use_catalog_notices(compiled)
+        notices = _use_catalog_notices(compiled) + monitor_notices(compiled.monitors)
         certified = sorted(
             {
                 f"{full_name} ({value})"
@@ -2489,6 +2555,11 @@ def plan_notices(self, contract: Mapping[str, Any]) -> List[str]:
             )
         return notices
 
+    def apply_failure_hints(self, address: str, detail: str = "") -> List[str]:
+        """The way on when ``tofu apply`` failed on ``address``: a data quality monitor
+        Databricks refused (``databricks_quality.failure_hints``); ``[]`` otherwise."""
+        return failure_hints(address)
+
     def explain_plan_failure(
         self,
         output: str,
@@ -2692,6 +2763,20 @@ def _job_actions(jobs: Sequence[DbtJob]) -> List[Dict[str, Any]]:
     return actions
 
 
+def _monitor_actions(monitors: Sequence[SchemaMonitor]) -> List[Dict[str, Any]]:
+    """One plan action per schema's data quality monitor, keyed by the schema it watches."""
+    return [
+        {
+            "op": "databricks.quality_monitor",
+            "resource_type": QUALITY_MONITOR,
+            "resource_id": monitor.full_name,
+            "monitored": list(monitor.monitored),
+            "excluded": list(monitor.excluded),
+        }
+        for monitor in monitors
+    ]
+
+
 def planned_objects(contract: Mapping[str, Any]) -> List[Dict[str, Any]]:
     """The objects the module manages, as plan actions, for the native planner.
 
@@ -2758,6 +2843,7 @@ def planned_objects(contract: Mapping[str, Any]) -> List[Dict[str, Any]]:
                 "tag_value": tag_value,
             }
         )
+    actions.extend(_monitor_actions(compiled.monitors))
     actions.extend(_job_actions(compiled.jobs))
     unique: Dict[Tuple[str, str], Dict[str, Any]] = {}
     for action in actions: