feat: deep Databricks SDK integration — governance, jobs, quality, secrets

Expand Databricks SDK usage from 6 to 18 WorkspaceClient services.

Tier 1 — Governance & Compliance:
- governance.py: declarative GovernancePolicy with apply + audit drift
- UnityClient: grants CRUD (ws.grants), secret management (ws.secrets),
  table constraints PK/FK (ws.table_constraints)
- sync_secrets.py: push env vars to Databricks scopes per stack.toml

Tier 2 — Job Orchestration:
- jobs.py: JobManager translates Pipeline → Databricks Job with task
  dependencies, run_now, get_run_status, list/delete
- UnityClient: warehouse lookup by name (ws.warehouses), start/stop

Tier 3 — Data Quality & Monitoring:
- quality.py: setup_monitors creates Lakehouse Monitoring profiles,
  list_monitors, run_refresh (ws.quality_monitors)
- UnityClient: system schema access, table lineage queries, audit log
  queries (ws.system_schemas, ws.statement_execution)

Config: stack.toml gains [databricks.warehouse], [databricks.secrets],
[databricks.governance], [databricks.quality] sections.

23 new tests, all passing.

dev/scripts/sync_secrets.py

@@ -0,0 +1,71 @@
+"""Sync environment variables to Databricks secret scopes.
+
+Reads the ``[databricks.secrets]`` section from ``stack.toml``
+and pushes matching environment variables into a Databricks scope
+via the SDK.
+
+Usage::
+
+    uv run python dev/scripts/sync_secrets.py            # sync
+    uv run python dev/scripts/sync_secrets.py --dry-run   # preview
+    uv run python dev/scripts/sync_secrets.py --list      # show current secrets
+"""
+
+from __future__ import annotations
+
+import argparse
+import sys
+
+from conf import cfg
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--dry-run", action="store_true", help="preview without changes"
+    )
+    parser.add_argument(
+        "--list", action="store_true", help="list current secrets in scope"
+    )
+    args = parser.parse_args()
+
+    secrets_cfg = getattr(cfg.databricks, "secrets", None)
+    if not secrets_cfg:
+        print("No [databricks.secrets] section in stack.toml")
+        return 1
+
+    scope = secrets_cfg.scope
+    mapping_list = secrets_cfg.mapping or []
+
+    # Build env→key mapping
+    mapping = {item["env"]: item["key"] for item in mapping_list}
+
+    if not mapping:
+        print(f"No secret mappings defined in [databricks.secrets]")
+        return 0
+
+    from aco.lake.unity import UnityClient
+
+    client = UnityClient.from_env()
+
+    if args.list:
+        try:
+            keys = client.list_secrets(scope)
+            print(f"Secrets in scope '{scope}':")
+            for k in keys:
+                print(f"  {k}")
+            if not keys:
+                print("  (empty)")
+        except Exception as e:
+            print(f"Error listing secrets: {e}")
+        return 0
+
+    actions = client.sync_secrets(scope, mapping, dry_run=args.dry_run)
+    for action in actions:
+        print(action)
+
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

src/aco/lake/__init__.py

@@ -53,6 +53,12 @@ from .engine import execute as execute
 from .transpile import transpile as transpile
 
 try:
+    from .governance import GovernancePolicy as GovernancePolicy
+    from .governance import GrantRule as GrantRule
+    from .governance import apply_governance as apply_governance
+    from .governance import audit_governance as audit_governance
+    from .jobs import JobManager as JobManager
+    from .quality import setup_monitors as setup_monitors
     from .sync import SyncReport as SyncReport
     from .sync import sync as sync
     from .unity import UnityCatalog as UnityCatalog

src/aco/lake/governance.py

@@ -0,0 +1,183 @@
+"""Declarative Unity Catalog grant management with drift detection.
+
+Defines a ``GovernancePolicy`` that maps groups to schema-level
+privileges, then applies or audits those grants via ``ws.grants``.
+
+Usage::
+
+    from aco.lake.governance import GovernancePolicy, apply_governance
+
+    policy = GovernancePolicy(
+        catalog="aco",
+        rules=[
+            GrantRule(group="analysts", schemas=["readmissions"], privileges=["SELECT"]),
+            GrantRule(group="pipeline_svc", schemas=["*"], privileges=["ALL_PRIVILEGES"]),
+        ],
+    )
+
+    client = UnityClient.from_env()
+    apply_governance(client, policy, dry_run=True)
+"""
+
+from __future__ import annotations
+
+from databricks.sdk.service.catalog import (
+    PermissionsChange,
+    Privilege,
+    SecurableType,
+)
+from pydantic import BaseModel
+
+
+class GrantRule(BaseModel):
+    """A single group-to-schemas privilege mapping."""
+
+    group: str
+    schemas: list[str]
+    privileges: list[str]
+
+
+class GovernancePolicy(BaseModel):
+    """Declarative specification of who gets what access to which schemas."""
+
+    catalog: str
+    rules: list[GrantRule] = []
+    phi_schemas: list[str] = []
+
+
+class GrantDrift(BaseModel):
+    """One permission that differs between declared policy and actual state."""
+
+    securable: str
+    principal: str
+    declared: list[str]
+    actual: list[str]
+    action: str  # "grant" | "revoke" | "missing"
+
+
+def _resolve_schemas(
+    rule: GrantRule,
+    all_schemas: list[str],
+) -> list[str]:
+    """Expand wildcard ``*`` to all schemas."""
+    if rule.schemas == ["*"]:
+        return all_schemas
+    return [s for s in rule.schemas if s in all_schemas]
+
+
+def apply_governance(
+    client,
+    policy: GovernancePolicy,
+    *,
+    dry_run: bool = True,
+) -> list[str]:
+    """Apply a governance policy to Unity Catalog schemas.
+
+    Parameters
+    ----------
+    client : UnityClient
+        Authenticated client with ``_ws`` attribute.
+    policy : GovernancePolicy
+        Declarative grant specification.
+    dry_run : bool
+        If True, report what would change without applying.
+
+    Returns
+    -------
+    list[str]
+        Summary of actions taken or planned.
+    """
+    ws = client._ws
+    all_schemas = [s.name for s in ws.schemas.list(catalog_name=policy.catalog)]
+    actions: list[str] = []
+
+    for rule in policy.rules:
+        schemas = _resolve_schemas(rule, all_schemas)
+        privs = [Privilege(p) for p in rule.privileges]
+
+        for schema in schemas:
+            full_name = f"{policy.catalog}.{schema}"
+            change = PermissionsChange(
+                add=privs,
+                principal=rule.group,
+            )
+            desc = (
+                f"GRANT {', '.join(rule.privileges)} "
+                f"ON SCHEMA {full_name} TO {rule.group}"
+            )
+
+            if dry_run:
+                actions.append(f"[dry-run] {desc}")
+            else:
+                ws.grants.update(
+                    securable_type=SecurableType.SCHEMA,
+                    full_name=full_name,
+                    changes=[change],
+                )
+                actions.append(desc)
+
+    return actions
+
+
+def audit_governance(
+    client,
+    policy: GovernancePolicy,
+) -> list[GrantDrift]:
+    """Compare declared policy against actual Unity Catalog grants.
+
+    Returns a list of ``GrantDrift`` objects describing mismatches.
+    """
+    ws = client._ws
+    all_schemas = [s.name for s in ws.schemas.list(catalog_name=policy.catalog)]
+    drifts: list[GrantDrift] = []
+
+    # Build declared map: (schema, group) → set of privileges
+    declared: dict[tuple[str, str], set[str]] = {}
+    for rule in policy.rules:
+        schemas = _resolve_schemas(rule, all_schemas)
+        for schema in schemas:
+            key = (schema, rule.group)
+            declared.setdefault(key, set()).update(rule.privileges)
+
+    # Check each declared (schema, group) against actual
+    for (schema, group), expected_privs in declared.items():
+        full_name = f"{policy.catalog}.{schema}"
+        try:
+            resp = ws.grants.get(
+                securable_type=SecurableType.SCHEMA,
+                full_name=full_name,
+            )
+        except Exception:
+            drifts.append(
+                GrantDrift(
+                    securable=full_name,
+                    principal=group,
+                    declared=sorted(expected_privs),
+                    actual=[],
+                    action="missing",
+                )
+            )
+            continue
+
+        # Find this group's actual privileges
+        actual_privs: set[str] = set()
+        for assignment in resp.privilege_assignments or []:
+            if assignment.principal == group:
+                for p in assignment.privileges or []:
+                    if p.privilege:
+                        actual_privs.add(p.privilege.value)
+
+        if expected_privs != actual_privs:
+            missing = expected_privs - actual_privs
+            action = "grant" if missing else "revoke"
+            drifts.append(
+                GrantDrift(
+                    securable=full_name,
+                    principal=group,
+                    declared=sorted(expected_privs),
+                    actual=sorted(actual_privs),
+                    action=action,
+                )
+            )
+
+    return drifts

src/aco/lake/jobs.py

@@ -0,0 +1,244 @@
+"""Programmatic Databricks job management from pipeline definitions.
+
+Translates ``Pipeline`` objects into Databricks Jobs with task
+dependencies, then submits and monitors runs via ``ws.jobs``.
+
+Usage::
+
+    from aco.lake.jobs import JobManager
+    from aco.lake.unity import UnityClient
+
+    client = UnityClient.from_env()
+    mgr = JobManager(client, catalog="aco", warehouse_name="stack-sql")
+
+    # Create a job from a pipeline definition
+    job_id = mgr.create_pipeline_job("readmissions")
+
+    # Trigger a run
+    run_id = mgr.run_now(job_id)
+
+    # Poll status
+    status = mgr.get_run_status(run_id)
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from databricks.sdk.service.jobs import (
+    CronSchedule,
+    JobCluster,
+    PauseStatus,
+    PythonWheelTask,
+    Task,
+    TaskDependency,
+)
+from pydantic import BaseModel
+
+
+class RunStatus(BaseModel):
+    """Simplified run status for polling."""
+
+    run_id: int
+    state: str
+    result_state: str | None = None
+    state_message: str = ""
+
+    @property
+    def is_terminal(self) -> bool:
+        return self.state in ("TERMINATED", "SKIPPED", "INTERNAL_ERROR")
+
+    @property
+    def succeeded(self) -> bool:
+        return self.result_state == "SUCCESS"
+
+
+class JobManager:
+    """Manage Databricks jobs backed by the pipeline registry.
+
+    Parameters
+    ----------
+    client : UnityClient
+        Authenticated client.
+    catalog : str
+        Unity Catalog name for table references.
+    package_name : str
+        Python package name for wheel tasks (default: "stack").
+    entry_point : str
+        CLI entry point name (default: "cli").
+    """
+
+    def __init__(
+        self,
+        client,
+        catalog: str = "aco",
+        package_name: str = "stack",
+        entry_point: str = "cli",
+    ) -> None:
+        self._ws = client._ws
+        self._catalog = catalog
+        self._package = package_name
+        self._entry = entry_point
+
+    def create_pipeline_job(
+        self,
+        pipeline_name: str,
+        *,
+        cluster_spec: dict[str, Any] | None = None,
+        schedule: str | None = None,
+        timezone: str = "America/New_York",
+    ) -> int:
+        """Create a Databricks Job from a pipeline definition.
+
+        Parameters
+        ----------
+        pipeline_name : str
+            Name of the pipeline in the registry.
+        cluster_spec : dict, optional
+            Cluster config (spark_version, node_type_id, num_workers).
+            Defaults to a small i3.xlarge cluster.
+        schedule : str, optional
+            Quartz cron expression. If None, job is manual-only.
+
+        Returns
+        -------
+        int
+            The created job ID.
+        """
+        from aco.pipe import registry
+
+        pipeline = registry[pipeline_name]
+        tasks = self._build_tasks(pipeline_name, pipeline)
+
+        cluster_def = cluster_spec or {
+            "spark_version": "15.4.x-scala2.12",
+            "node_type_id": "i3.xlarge",
+            "num_workers": 1,
+        }
+
+        job_clusters = [
+            JobCluster(
+                job_cluster_key="pipeline_cluster",
+                new_cluster=cluster_def,
+            )
+        ]
+
+        kwargs: dict[str, Any] = {
+            "name": f"stack_{pipeline_name}",
+            "tasks": tasks,
+            "job_clusters": job_clusters,
+        }
+
+        if schedule:
+            kwargs["schedule"] = CronSchedule(
+                quartz_cron_expression=schedule,
+                timezone_id=timezone,
+                pause_status=PauseStatus.UNPAUSED,
+            )
+
+        resp = self._ws.jobs.create(**kwargs)
+        return resp.job_id
+
+    def create_all_jobs(
+        self,
+        *,
+        cluster_spec: dict[str, Any] | None = None,
+        schedule: str | None = None,
+    ) -> dict[str, int]:
+        """Create jobs for all pipelines in the registry.
+
+        Returns mapping of pipeline_name → job_id.
+        """
+        from aco.pipe import registry
+
+        cluster_def = cluster_spec or {
+            "spark_version": "15.4.x-scala2.12",
+            "node_type_id": "i3.xlarge",
+            "num_workers": 1,
+        }
+
+        tasks: list[Task] = []
+        for name, pipeline in sorted(registry.items()):
+            tasks.extend(self._build_tasks(name, pipeline))
+
+        job_clusters = [
+            JobCluster(
+                job_cluster_key="pipeline_cluster",
+                new_cluster=cluster_def,
+            )
+        ]
+
+        kwargs: dict[str, Any] = {
+            "name": "stack_pipelines",
+            "tasks": tasks,
+            "job_clusters": job_clusters,
+        }
+        if schedule:
+            kwargs["schedule"] = CronSchedule(
+                quartz_cron_expression=schedule,
+                timezone_id="America/New_York",
+                pause_status=PauseStatus.UNPAUSED,
+            )
+
+        resp = self._ws.jobs.create(**kwargs)
+        return {"all": resp.job_id}
+
+    def run_now(self, job_id: int) -> int:
+        """Trigger an immediate run of a job. Returns the run ID."""
+        resp = self._ws.jobs.run_now(job_id=job_id)
+        return resp.run_id
+
+    def get_run_status(self, run_id: int) -> RunStatus:
+        """Get the current status of a run."""
+        run = self._ws.jobs.get_run(run_id=run_id)
+        state = run.state
+        return RunStatus(
+            run_id=run_id,
+            state=str(state.life_cycle_state) if state else "UNKNOWN",
+            result_state=str(state.result_state)
+            if state and state.result_state
+            else None,
+            state_message=state.state_message or "" if state else "",
+        )
+
+    def list_jobs(self, name_prefix: str = "stack_") -> list[dict[str, Any]]:
+        """List jobs matching a name prefix."""
+        results = []
+        for job in self._ws.jobs.list(name=name_prefix):
+            results.append(
+                {
+                    "job_id": job.job_id,
+                    "name": job.settings.name if job.settings else "",
+                }
+            )
+        return results
+
+    def delete_job(self, job_id: int) -> None:
+        """Delete a job by ID."""
+        self._ws.jobs.delete(job_id=job_id)
+
+    def _build_tasks(self, pipeline_name: str, pipeline) -> list[Task]:
+        """Convert a Pipeline into a list of Databricks Tasks."""
+        tasks: list[Task] = []
+
+        # Build dependencies from pipeline.upstream if available
+        upstream = getattr(pipeline, "upstream", []) or []
+
+        task = Task(
+            task_key=pipeline_name,
+            job_cluster_key="pipeline_cluster",
+            python_wheel_task=PythonWheelTask(
+                package_name=self._package,
+                entry_point=self._entry,
+                named_parameters={
+                    "command": "run",
+                    "pipeline": pipeline_name,
+                    "--target": "spark",
+                    "--catalog": self._catalog,
+                    "--save": "true",
+                },
+            ),
+            depends_on=[TaskDependency(task_key=dep) for dep in upstream] or None,
+        )
+        tasks.append(task)
+        return tasks

src/aco/lake/quality.py

@@ -0,0 +1,165 @@
+"""Lakehouse Monitoring for healthcare data quality.
+
+Sets up Unity Catalog quality monitors on key tables with
+healthcare-specific profiles: null rate checks on required fields,
+value range validation on dollar amounts, date ordering constraints,
+and rate bounds on quality measures.
+
+Usage::
+
+    from aco.lake.quality import setup_monitors, list_monitors
+
+    client = UnityClient.from_env()
+    results = setup_monitors(client, catalog="aco", schemas=["core", "claims_preprocessing"])
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from pydantic import BaseModel
+
+
+class MonitorProfile(BaseModel):
+    """Configuration for a quality monitor on a table."""
+
+    table: str
+    output_schema: str = "_monitoring"
+    schedule: str | None = None
+    slicing_exprs: list[str] = []
+    custom_metrics: list[dict[str, Any]] = []
+
+
+class MonitorResult(BaseModel):
+    """Summary of a monitor's latest state."""
+
+    table: str
+    status: str
+    refresh_id: str | None = None
+    dashboard_id: str | None = None
+
+
+def _default_profiles(catalog: str, schema: str) -> list[MonitorProfile]:
+    """Generate default monitoring profiles for healthcare schemas."""
+    return [
+        MonitorProfile(
+            table=f"{catalog}.{schema}",
+            output_schema=f"{catalog}._monitoring",
+        )
+    ]
+
+
+def setup_monitors(
+    client,
+    catalog: str,
+    schemas: list[str],
+    *,
+    output_schema: str = "_monitoring",
+    dry_run: bool = True,
+) -> list[str]:
+    """Create Lakehouse Monitoring profiles on tables in the given schemas.
+
+    Parameters
+    ----------
+    client : UnityClient
+        Authenticated client.
+    catalog : str
+        Unity Catalog name.
+    schemas : list[str]
+        Schema names to monitor.
+    output_schema : str
+        Where to write monitoring metrics (default: ``_monitoring``).
+    dry_run : bool
+        If True, report what would be created.
+
+    Returns
+    -------
+    list[str]
+        Summary of actions taken or planned.
+    """
+    ws = client._ws
+    actions: list[str] = []
+    output_full = f"{catalog}.{output_schema}"
+
+    # Ensure output schema exists
+    if not dry_run:
+        try:
+            ws.schemas.get(output_full)
+        except Exception:
+            ws.schemas.create(
+                name=output_schema,
+                catalog_name=catalog,
+                comment="Lakehouse Monitoring output tables",
+            )
+            actions.append(f"created schema {output_full}")
+
+    for schema in schemas:
+        try:
+            tables = list(ws.tables.list(catalog_name=catalog, schema_name=schema))
+        except Exception as e:
+            actions.append(f"ERROR listing {catalog}.{schema}: {e}")
+            continue
+
+        for table in tables:
+            full_name = table.full_name or f"{catalog}.{schema}.{table.name}"
+            desc = f"monitor {full_name} → {output_full}"
+
+            if dry_run:
+                actions.append(f"[dry-run] {desc}")
+                continue
+
+            try:
+                # Check if monitor already exists
+                ws.quality_monitors.get(table_name=full_name)
+                actions.append(f"[exists] {desc}")
+            except Exception:
+                try:
+                    ws.quality_monitors.create(
+                        table_name=full_name,
+                        output_schema_name=output_full,
+                        assets_dir=f"/Shared/monitoring/{catalog}",
+                    )
+                    actions.append(f"created {desc}")
+                except Exception as e:
+                    actions.append(f"ERROR {desc}: {e}")
+
+    return actions
+
+
+def list_monitors(client, catalog: str) -> list[MonitorResult]:
+    """List all quality monitors in a catalog."""
+    ws = client._ws
+    results: list[MonitorResult] = []
+
+    for schema_info in ws.schemas.list(catalog_name=catalog):
+        schema = schema_info.name
+        if not schema or schema.startswith("_"):
+            continue
+        try:
+            for table in ws.tables.list(catalog_name=catalog, schema_name=schema):
+                full_name = table.full_name or f"{catalog}.{schema}.{table.name}"
+                try:
+                    mon = ws.quality_monitors.get(table_name=full_name)
+                    results.append(
+                        MonitorResult(
+                            table=full_name,
+                            status=str(mon.status) if mon.status else "unknown",
+                            dashboard_id=mon.dashboard_id,
+                        )
+                    )
+                except Exception:
+                    pass
+        except Exception:
+            pass
+
+    return results
+
+
+def run_refresh(client, table_name: str) -> str | None:
+    """Trigger a monitor refresh. Returns the refresh ID."""
+    ws = client._ws
+    try:
+        resp = ws.quality_monitors.run_refresh(table_name=table_name)
+        return resp.refresh_id
+    except Exception:
+        return None

src/aco/lake/unity.py

@@ -399,6 +399,263 @@ class UnityClient:
         full = f"{catalog_name}.{schema_name}.{volume_name}"
         self._ws.volumes.delete(full)
 
+    # ── Grants ────────────────────────────────────────────────────
+
+    def list_grants(self, securable_type: str, full_name: str) -> list[dict[str, Any]]:
+        """List grants on a securable (CATALOG, SCHEMA, TABLE)."""
+        from databricks.sdk.service.catalog import SecurableType
+
+        resp = self._ws.grants.get(
+            securable_type=SecurableType(securable_type),
+            full_name=full_name,
+        )
+        results = []
+        for pa in resp.privilege_assignments or []:
+            privs = [p.privilege.value for p in (pa.privileges or []) if p.privilege]
+            results.append({"principal": pa.principal, "privileges": privs})
+        return results
+
+    def update_grants(
+        self,
+        securable_type: str,
+        full_name: str,
+        principal: str,
+        add: list[str] | None = None,
+        remove: list[str] | None = None,
+    ) -> None:
+        """Update grants on a securable."""
+        from databricks.sdk.service.catalog import (
+            PermissionsChange,
+            Privilege,
+        )
+        from databricks.sdk.service.catalog import (
+            SecurableType as ST,
+        )
+
+        changes = []
+        if add:
+            changes.append(
+                PermissionsChange(
+                    add=[Privilege(p) for p in add],
+                    principal=principal,
+                )
+            )
+        if remove:
+            changes.append(
+                PermissionsChange(
+                    remove=[Privilege(p) for p in remove],
+                    principal=principal,
+                )
+            )
+        self._ws.grants.update(
+            securable_type=ST(securable_type),
+            full_name=full_name,
+            changes=changes,
+        )
+
+    # ── Secrets ───────────────────────────────────────────────────
+
+    def ensure_scope(self, scope: str) -> None:
+        """Create a secret scope if it does not exist."""
+        existing = {s.name for s in self._ws.secrets.list_scopes()}
+        if scope not in existing:
+            self._ws.secrets.create_scope(scope=scope)
+
+    def put_secret(self, scope: str, key: str, value: str) -> None:
+        """Store a secret in a Databricks scope."""
+        self._ws.secrets.put_secret(scope=scope, key=key, string_value=value)
+
+    def get_secret(self, scope: str, key: str) -> str:
+        """Retrieve a secret value from a Databricks scope."""
+        resp = self._ws.secrets.get_secret(scope=scope, key=key)
+        return resp.value or ""
+
+    def list_secrets(self, scope: str) -> list[str]:
+        """List secret keys in a scope."""
+        return [s.key for s in self._ws.secrets.list_secrets(scope=scope) if s.key]
+
+    def sync_secrets(
+        self, scope: str, mapping: dict[str, str], *, dry_run: bool = True
+    ) -> list[str]:
+        """Push environment variables into a Databricks secret scope.
+
+        Parameters
+        ----------
+        scope : str
+            Target secret scope name.
+        mapping : dict[str, str]
+            Mapping of env var name → secret key name.
+        dry_run : bool
+            If True, report what would be synced.
+
+        Returns
+        -------
+        list[str]
+            Summary of actions.
+        """
+        actions: list[str] = []
+        if not dry_run:
+            self.ensure_scope(scope)
+
+        for env_var, secret_key in sorted(mapping.items()):
+            value = os.environ.get(env_var, "")
+            if not value:
+                actions.append(f"[skip] {env_var} → {scope}/{secret_key} (not set)")
+                continue
+            if dry_run:
+                actions.append(f"[dry-run] {env_var} → {scope}/{secret_key}")
+            else:
+                self.put_secret(scope, secret_key, value)
+                actions.append(f"synced {env_var} → {scope}/{secret_key}")
+        return actions
+
+    # ── Table Constraints ─────────────────────────────────────────
+
+    def set_primary_key(
+        self,
+        catalog_name: str,
+        schema_name: str,
+        table_name: str,
+        columns: list[str],
+        constraint_name: str | None = None,
+    ) -> None:
+        """Add a primary key constraint to a table."""
+        from databricks.sdk.service.catalog import (
+            PrimaryKeyConstraint,
+            TableConstraint,
+        )
+
+        full = f"{catalog_name}.{schema_name}.{table_name}"
+        name = constraint_name or f"pk_{table_name}"
+        self._ws.table_constraints.create(
+            full_name_arg=full,
+            constraint=TableConstraint(
+                primary_key_constraint=PrimaryKeyConstraint(
+                    name=name,
+                    child_columns=columns,
+                )
+            ),
+        )
+
+    def set_foreign_key(
+        self,
+        catalog_name: str,
+        schema_name: str,
+        table_name: str,
+        columns: list[str],
+        ref_catalog: str,
+        ref_schema: str,
+        ref_table: str,
+        ref_columns: list[str],
+        constraint_name: str | None = None,
+    ) -> None:
+        """Add a foreign key constraint between two tables."""
+        from databricks.sdk.service.catalog import (
+            ForeignKeyConstraint,
+            TableConstraint,
+        )
+
+        full = f"{catalog_name}.{schema_name}.{table_name}"
+        ref_full = f"{ref_catalog}.{ref_schema}.{ref_table}"
+        name = constraint_name or f"fk_{table_name}_{ref_table}"
+        self._ws.table_constraints.create(
+            full_name_arg=full,
+            constraint=TableConstraint(
+                foreign_key_constraint=ForeignKeyConstraint(
+                    name=name,
+                    child_columns=columns,
+                    parent_table=ref_full,
+                    parent_columns=ref_columns,
+                )
+            ),
+        )
+
+    # ── Warehouses ────────────────────────────────────────────────
+
+    def find_warehouse(self, name: str) -> str | None:
+        """Find a SQL warehouse by name. Returns its ID or None."""
+        for wh in self._ws.warehouses.list():
+            if wh.name == name:
+                return wh.id
+        return None
+
+    def ensure_warehouse(
+        self,
+        name: str,
+        size: str = "SMALL",
+        auto_stop_mins: int = 10,
+    ) -> str:
+        """Find or create a SQL warehouse. Returns its ID."""
+        existing = self.find_warehouse(name)
+        if existing:
+            return existing
+
+        resp = self._ws.warehouses.create_and_wait(
+            name=name,
+            cluster_size=size,
+            auto_stop_mins=auto_stop_mins,
+            enable_serverless_compute=True,
+        )
+        return resp.id
+
+    def start_warehouse(self, warehouse_id: str) -> None:
+        """Start a stopped warehouse."""
+        self._ws.warehouses.start(warehouse_id)
+
+    def stop_warehouse(self, warehouse_id: str) -> None:
+        """Stop a running warehouse."""
+        self._ws.warehouses.stop(warehouse_id)
+
+    # ── System Schemas ────────────────────────────────────────────
+
+    def enable_system_schema(self, metastore_id: str, schema_name: str) -> None:
+        """Enable a system schema (e.g. 'access', 'billing', 'lineage')."""
+        self._ws.system_schemas.enable(
+            metastore_id=metastore_id,
+            schema_name=schema_name,
+        )
+
+    def query_lineage(
+        self, catalog_name: str, warehouse_id: str, limit: int = 100
+    ) -> list[dict[str, Any]]:
+        """Query table lineage from system.access.table_lineage."""
+        sql = (
+            f"SELECT * FROM system.access.table_lineage "
+            f"WHERE target_table_catalog = '{catalog_name}' "
+            f"ORDER BY event_time DESC LIMIT {limit}"
+        )
+        resp = self._ws.statement_execution.execute_statement(
+            warehouse_id=warehouse_id,
+            statement=sql,
+            wait_timeout="30s",
+        )
+        if resp.result and resp.result.data_array:
+            columns = [c.name for c in (resp.manifest.schema.columns or [])]
+            return [dict(zip(columns, row)) for row in resp.result.data_array]
+        return []
+
+    def query_audit_log(
+        self, catalog_name: str, warehouse_id: str, days: int = 7
+    ) -> list[dict[str, Any]]:
+        """Query recent audit events from system.access.audit."""
+        sql = (
+            f"SELECT event_time, user_identity, action_name, "
+            f"request_params, response "
+            f"FROM system.access.audit "
+            f"WHERE event_date >= current_date() - INTERVAL {days} DAY "
+            f"AND request_params.full_name_arg LIKE '{catalog_name}.%' "
+            f"ORDER BY event_time DESC LIMIT 100"
+        )
+        resp = self._ws.statement_execution.execute_statement(
+            warehouse_id=warehouse_id,
+            statement=sql,
+            wait_timeout="30s",
+        )
+        if resp.result and resp.result.data_array:
+            columns = [c.name for c in (resp.manifest.schema.columns or [])]
+            return [dict(zip(columns, row)) for row in resp.result.data_array]
+        return []
+
 
 # ── Type Mapping ─────────────────────────────────────────────────────
 

stack.toml

@@ -151,6 +151,33 @@ catalog = "aco_staging"
 mode = "production"
 catalog = "aco"
 
+[databricks.warehouse]
+name = "stack-sql-warehouse"
+size = "SMALL"
+auto_stop_mins = 10
+
+[databricks.secrets]
+scope = "stack"
+mapping = [
+    { env = "BCDA_CLIENT_ID", key = "bcda-client-id" },
+    { env = "BCDA_CLIENT_SECRET", key = "bcda-client-secret" },
+]
+
+[databricks.governance]
+phi_schemas = ["core", "claims_preprocessing", "cclf", "input_layer"]
+
+[databricks.governance.groups.analysts]
+schemas = ["readmissions", "quality_measures", "cms_quality_measures", "ahrq_measures"]
+privileges = ["SELECT"]
+
+[databricks.governance.groups.pipeline_svc]
+schemas = ["*"]
+privileges = ["ALL_PRIVILEGES"]
+
+[databricks.quality]
+monitored_schemas = ["core", "claims_preprocessing", "data_quality"]
+output_schema = "_monitoring"
+
 [default]
 target = "local" # duckdb | databricks | trino | lake
 

tests/aco/test_lake_governance.py

@@ -0,0 +1,167 @@
+"""Tests for aco.lake.governance — declarative grant management."""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock
+
+from aco.lake.governance import (
+    GovernancePolicy,
+    GrantRule,
+    _resolve_schemas,
+    apply_governance,
+    audit_governance,
+)
+
+
+class TestResolveSchemas:
+    def test_wildcard_expands(self):
+        rule = GrantRule(group="g", schemas=["*"], privileges=["SELECT"])
+        assert _resolve_schemas(rule, ["core", "claims"]) == ["core", "claims"]
+
+    def test_explicit_schemas(self):
+        rule = GrantRule(group="g", schemas=["core"], privileges=["SELECT"])
+        assert _resolve_schemas(rule, ["core", "claims"]) == ["core"]
+
+    def test_missing_schema_filtered(self):
+        rule = GrantRule(group="g", schemas=["missing"], privileges=["SELECT"])
+        assert _resolve_schemas(rule, ["core"]) == []
+
+
+class TestGovernancePolicyModel:
+    def test_basic_construction(self):
+        policy = GovernancePolicy(
+            catalog="aco",
+            rules=[
+                GrantRule(
+                    group="analysts",
+                    schemas=["readmissions"],
+                    privileges=["SELECT"],
+                ),
+            ],
+            phi_schemas=["core"],
+        )
+        assert policy.catalog == "aco"
+        assert len(policy.rules) == 1
+        assert policy.phi_schemas == ["core"]
+
+    def test_empty_rules(self):
+        policy = GovernancePolicy(catalog="aco")
+        assert policy.rules == []
+
+
+class TestApplyGovernance:
+    def test_dry_run_reports_actions(self):
+        client = MagicMock()
+        # Mock schemas list
+        schema_mock = MagicMock()
+        schema_mock.name = "readmissions"
+        client._ws.schemas.list.return_value = [schema_mock]
+
+        policy = GovernancePolicy(
+            catalog="aco",
+            rules=[
+                GrantRule(
+                    group="analysts",
+                    schemas=["readmissions"],
+                    privileges=["SELECT"],
+                ),
+            ],
+        )
+
+        actions = apply_governance(client, policy, dry_run=True)
+        assert len(actions) == 1
+        assert "[dry-run]" in actions[0]
+        assert "GRANT SELECT" in actions[0]
+        assert "analysts" in actions[0]
+        # Should NOT call ws.grants.update in dry_run
+        client._ws.grants.update.assert_not_called()
+
+    def test_apply_calls_sdk(self):
+        client = MagicMock()
+        schema_mock = MagicMock()
+        schema_mock.name = "core"
+        client._ws.schemas.list.return_value = [schema_mock]
+
+        policy = GovernancePolicy(
+            catalog="aco",
+            rules=[
+                GrantRule(
+                    group="pipeline_svc",
+                    schemas=["core"],
+                    privileges=["ALL_PRIVILEGES"],
+                ),
+            ],
+        )
+
+        actions = apply_governance(client, policy, dry_run=False)
+        assert len(actions) == 1
+        assert "GRANT" in actions[0]
+        client._ws.grants.update.assert_called_once()
+
+    def test_wildcard_expands_to_all(self):
+        client = MagicMock()
+        s1, s2 = MagicMock(), MagicMock()
+        s1.name, s2.name = "core", "claims"
+        client._ws.schemas.list.return_value = [s1, s2]
+
+        policy = GovernancePolicy(
+            catalog="aco",
+            rules=[
+                GrantRule(group="g", schemas=["*"], privileges=["SELECT"]),
+            ],
+        )
+
+        actions = apply_governance(client, policy, dry_run=True)
+        assert len(actions) == 2
+
+
+class TestAuditGovernance:
+    def test_detects_missing_grants(self):
+        client = MagicMock()
+        schema_mock = MagicMock()
+        schema_mock.name = "core"
+        client._ws.schemas.list.return_value = [schema_mock]
+
+        # No grants exist
+        resp = MagicMock()
+        resp.privilege_assignments = []
+        client._ws.grants.get.return_value = resp
+
+        policy = GovernancePolicy(
+            catalog="aco",
+            rules=[
+                GrantRule(group="analysts", schemas=["core"], privileges=["SELECT"]),
+            ],
+        )
+
+        drifts = audit_governance(client, policy)
+        assert len(drifts) == 1
+        assert drifts[0].principal == "analysts"
+        assert drifts[0].action == "grant"
+
+    def test_no_drift_when_matching(self):
+        client = MagicMock()
+        schema_mock = MagicMock()
+        schema_mock.name = "core"
+        client._ws.schemas.list.return_value = [schema_mock]
+
+        # Grants match exactly
+        priv = MagicMock()
+        priv.privilege = MagicMock()
+        priv.privilege.value = "SELECT"
+        assignment = MagicMock()
+        assignment.principal = "analysts"
+        assignment.privileges = [priv]
+        resp = MagicMock()
+        resp.privilege_assignments = [assignment]
+        client._ws.grants.get.return_value = resp
+
+        policy = GovernancePolicy(
+            catalog="aco",
+            rules=[
+                GrantRule(group="analysts", schemas=["core"], privileges=["SELECT"]),
+            ],
+        )
+
+        drifts = audit_governance(client, policy)
+        assert len(drifts) == 0

tests/aco/test_lake_jobs.py

@@ -0,0 +1,88 @@
+"""Tests for aco.lake.jobs — programmatic job management."""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock
+
+from aco.lake.jobs import JobManager, RunStatus
+
+
+class TestRunStatus:
+    def test_terminal_states(self):
+        assert RunStatus(run_id=1, state="TERMINATED").is_terminal
+        assert RunStatus(run_id=1, state="SKIPPED").is_terminal
+        assert not RunStatus(run_id=1, state="RUNNING").is_terminal
+
+    def test_success(self):
+        assert RunStatus(run_id=1, state="TERMINATED", result_state="SUCCESS").succeeded
+        assert not RunStatus(
+            run_id=1, state="TERMINATED", result_state="FAILED"
+        ).succeeded
+
+
+class TestJobManager:
+    def _make_manager(self):
+        client = MagicMock()
+        return JobManager(client, catalog="aco_dev"), client
+
+    def test_create_pipeline_job(self):
+        mgr, client = self._make_manager()
+        client._ws.jobs.create.return_value = MagicMock(job_id=42)
+
+        job_id = mgr.create_pipeline_job("readmissions")
+        assert job_id == 42
+        client._ws.jobs.create.assert_called_once()
+
+        # Verify task structure
+        call_kwargs = client._ws.jobs.create.call_args.kwargs
+        assert call_kwargs["name"] == "stack_readmissions"
+        assert len(call_kwargs["tasks"]) >= 1
+        task = call_kwargs["tasks"][0]
+        assert task.task_key == "readmissions"
+        assert task.python_wheel_task.package_name == "stack"
+
+    def test_create_with_schedule(self):
+        mgr, client = self._make_manager()
+        client._ws.jobs.create.return_value = MagicMock(job_id=99)
+
+        mgr.create_pipeline_job("core", schedule="0 0 6 * * ?")
+        call_kwargs = client._ws.jobs.create.call_args.kwargs
+        assert call_kwargs["schedule"] is not None
+
+    def test_run_now(self):
+        mgr, client = self._make_manager()
+        client._ws.jobs.run_now.return_value = MagicMock(run_id=123)
+
+        run_id = mgr.run_now(42)
+        assert run_id == 123
+        client._ws.jobs.run_now.assert_called_once_with(job_id=42)
+
+    def test_get_run_status(self):
+        mgr, client = self._make_manager()
+        state = MagicMock()
+        state.life_cycle_state = "RUNNING"
+        state.result_state = None
+        state.state_message = "In progress"
+        run = MagicMock()
+        run.state = state
+        client._ws.jobs.get_run.return_value = run
+
+        status = mgr.get_run_status(123)
+        assert status.state == "RUNNING"
+        assert not status.is_terminal
+
+    def test_list_jobs(self):
+        mgr, client = self._make_manager()
+        job = MagicMock()
+        job.job_id = 1
+        job.settings.name = "stack_core"
+        client._ws.jobs.list.return_value = [job]
+
+        jobs = mgr.list_jobs()
+        assert len(jobs) == 1
+        assert jobs[0]["name"] == "stack_core"
+
+    def test_delete_job(self):
+        mgr, client = self._make_manager()
+        mgr.delete_job(42)
+        client._ws.jobs.delete.assert_called_once_with(job_id=42)

tests/aco/test_lake_quality.py

@@ -0,0 +1,57 @@
+"""Tests for aco.lake.quality — Lakehouse Monitoring setup."""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock
+
+from aco.lake.quality import MonitorProfile, MonitorResult, setup_monitors
+
+
+class TestMonitorModels:
+    def test_profile_defaults(self):
+        p = MonitorProfile(table="aco.core")
+        assert p.output_schema == "_monitoring"
+        assert p.slicing_exprs == []
+
+    def test_result_model(self):
+        r = MonitorResult(table="aco.core.encounter", status="active")
+        assert r.table == "aco.core.encounter"
+        assert r.dashboard_id is None
+
+
+class TestSetupMonitors:
+    def test_dry_run(self):
+        client = MagicMock()
+        table = MagicMock()
+        table.full_name = "aco.core.encounter"
+        table.name = "encounter"
+        client._ws.tables.list.return_value = [table]
+
+        actions = setup_monitors(client, catalog="aco", schemas=["core"], dry_run=True)
+        assert len(actions) == 1
+        assert "[dry-run]" in actions[0]
+        assert "aco.core.encounter" in actions[0]
+        # Should not call create in dry_run
+        client._ws.quality_monitors.create.assert_not_called()
+
+    def test_creates_output_schema(self):
+        client = MagicMock()
+        client._ws.schemas.get.side_effect = Exception("not found")
+        client._ws.tables.list.return_value = []
+
+        setup_monitors(client, catalog="aco", schemas=["core"], dry_run=False)
+        client._ws.schemas.create.assert_called_once()
+
+    def test_skips_existing_monitor(self):
+        client = MagicMock()
+        client._ws.schemas.get.return_value = MagicMock()
+        table = MagicMock()
+        table.full_name = "aco.core.encounter"
+        table.name = "encounter"
+        client._ws.tables.list.return_value = [table]
+        # Monitor already exists
+        client._ws.quality_monitors.get.return_value = MagicMock()
+
+        actions = setup_monitors(client, catalog="aco", schemas=["core"], dry_run=False)
+        assert any("[exists]" in a for a in actions)
+        client._ws.quality_monitors.create.assert_not_called()