diff --git a/2026-04-workspace-maintenance-cleanup/README.md b/2026-04-workspace-maintenance-cleanup/README.md
new file mode 100644
index 0000000..e2ac14c
--- /dev/null
+++ b/2026-04-workspace-maintenance-cleanup/README.md
@@ -0,0 +1,49 @@
+# Databricks Workspace Cleanup
+
+Automated workspace maintenance using Databricks Asset Bundles — clean up unused jobs, dashboards, vector search indexes, and clusters with full audit logging and Lakeview dashboards.
+
+## Two Approaches
+
+1. **API-Driven Cleanup** — REST APIs scan and remove unused resources by metadata (last run date, status, source table existence)
+2. **System Table-Driven Cleanup** — Query `system.billing.usage`, `system.compute.clusters`, `system.lakeflow.job_run_timeline`, and `system.query.history` to find resources costing money but delivering no value
+
+Both approaches log every action (deleted, skipped, flagged) to a Delta table for auditability.
+
+## Structure
+
+```
+├── databricks.yml              # DAB bundle definition
+├── config/
+│   ├── config.yaml             # Cleanup toggles per environment
+│   └── thresholds.yaml         # Retention thresholds
+├── notebooks/
+│   ├── 00_cleanup_logger.py    # Structured logging module
+│   ├── 01_api_job_cleanup.py   # API: delete inactive jobs
+│   ├── 02_api_dashboard_cleanup.py  # API: remove stale dashboards
+│   ├── 03_api_vector_cleanup.py     # API: purge orphaned indexes
+│   ├── 04_system_table_analysis.py  # System tables: discover waste
+│   └── 05_system_table_cleanup.py   # System tables: act on flagged items
+└── dashboards/
+    └── cleanup_dashboard.sql   # Lakeview dashboard queries
+```
+
+## Quick Start
+
+```bash
+# Deploy to dev (dry run)
+databricks bundle deploy --target dev
+databricks bundle run cleanup_workflow --target dev
+
+# Review flagged items in the Lakeview dashboard, then:
+databricks bundle deploy --target prod
+```
+
+## Configuration
+
+Edit `config/config.yaml` to toggle cleanups per environment. Edit `config/thresholds.yaml` to set retention periods.
+
+Dev always runs in dry-run mode. Production executes deletions on a weekly Sunday 2 AM schedule.
+
+## Blog Post
+
+[Databricks Maintenance and Cleanup: Visualise, Clean, and Log with Asset Bundles](https://community.databricks.com/t5/technical-blog/databricks-maintenance-and-cleanup-visualise-clean-and-log-with/ba-p/135657)
diff --git a/2026-04-workspace-maintenance-cleanup/config/config.yaml b/2026-04-workspace-maintenance-cleanup/config/config.yaml
new file mode 100644
index 0000000..a1501dc
--- /dev/null
+++ b/2026-04-workspace-maintenance-cleanup/config/config.yaml
@@ -0,0 +1,20 @@
+dev:
+  job_cleanup: false
+  dashboard_cleanup: false
+  vector_cleanup: true
+  system_table_cleanup: true
+  dry_run: true
+
+stage:
+  job_cleanup: true
+  dashboard_cleanup: true
+  vector_cleanup: true
+  system_table_cleanup: true
+  dry_run: false
+
+prod:
+  job_cleanup: true
+  dashboard_cleanup: true
+  vector_cleanup: true
+  system_table_cleanup: true
+  dry_run: false
diff --git a/2026-04-workspace-maintenance-cleanup/config/thresholds.yaml b/2026-04-workspace-maintenance-cleanup/config/thresholds.yaml
new file mode 100644
index 0000000..246e2f4
--- /dev/null
+++ b/2026-04-workspace-maintenance-cleanup/config/thresholds.yaml
@@ -0,0 +1,5 @@
+job_inactive_days: 90
+dashboard_inactive_days: 60
+cluster_idle_hours: 48
+cost_anomaly_multiplier: 3.0
+min_dbu_threshold: 0.01
diff --git a/2026-04-workspace-maintenance-cleanup/dashboards/cleanup_dashboard.sql b/2026-04-workspace-maintenance-cleanup/dashboards/cleanup_dashboard.sql
new file mode 100644
index 0000000..a682b4f
--- /dev/null
+++ b/2026-04-workspace-maintenance-cleanup/dashboards/cleanup_dashboard.sql
@@ -0,0 +1,44 @@
+-- ============================================================
+-- Lakeview Dashboard Queries for Workspace Cleanup
+-- ============================================================
+
+-- 1. Cleanup Summary by Type and Action
+SELECT resource_type, action, COUNT(*) AS count, environment
+FROM maintenance.cleanup.cleanup_log
+WHERE timestamp >= DATEADD(DAY, -30, CURRENT_DATE())
+GROUP BY resource_type, action, environment
+ORDER BY count DESC;
+
+-- 2. Daily Cleanup Trend
+SELECT DATE(timestamp) AS cleanup_date, action, COUNT(*) AS count
+FROM maintenance.cleanup.cleanup_log
+WHERE timestamp >= DATEADD(DAY, -30, CURRENT_DATE())
+GROUP BY DATE(timestamp), action
+ORDER BY cleanup_date;
+
+-- 3. Top Wasted Resources by Cost
+SELECT resource_type, resource_name, reason,
+    get_json_object(details, '$.cost') AS estimated_cost,
+    owner, timestamp
+FROM maintenance.cleanup.cleanup_log
+WHERE action = 'FLAGGED'
+    AND timestamp >= DATEADD(DAY, -7, CURRENT_DATE())
+ORDER BY CAST(get_json_object(details, '$.cost') AS DOUBLE) DESC
+LIMIT 20;
+
+-- 4. Environment-Wise Retention
+SELECT environment, resource_type,
+    SUM(CASE WHEN action = 'DELETED' THEN 1 ELSE 0 END) AS deleted,
+    SUM(CASE WHEN action = 'SKIPPED' THEN 1 ELSE 0 END) AS retained,
+    SUM(CASE WHEN action = 'FLAGGED' THEN 1 ELSE 0 END) AS flagged,
+    SUM(CASE WHEN action = 'DRY_RUN' THEN 1 ELSE 0 END) AS dry_run
+FROM maintenance.cleanup.cleanup_log
+WHERE timestamp >= DATEADD(DAY, -30, CURRENT_DATE())
+GROUP BY environment, resource_type;
+
+-- 5. Full Audit Trail
+SELECT timestamp, environment, resource_type, resource_id,
+    resource_name, owner, action, reason, dry_run, details
+FROM maintenance.cleanup.cleanup_log
+ORDER BY timestamp DESC
+LIMIT 100;
diff --git a/2026-04-workspace-maintenance-cleanup/databricks.yml b/2026-04-workspace-maintenance-cleanup/databricks.yml
new file mode 100644
index 0000000..54bd79d
--- /dev/null
+++ b/2026-04-workspace-maintenance-cleanup/databricks.yml
@@ -0,0 +1,72 @@
+bundle:
+  name: workspace-cleanup
+
+workspace:
+  host: https://<your-workspace>.azuredatabricks.net
+
+variables:
+  environment:
+    default: dev
+
+resources:
+  jobs:
+    cleanup_workflow:
+      name: "[Maintenance] Workspace Cleanup"
+      schedule:
+        quartz_cron_expression: "0 0 2 ? * SUN"
+        timezone_id: "UTC"
+      parameters:
+        - name: environment
+          default: ${var.environment}
+      tasks:
+        - task_key: system_table_analysis
+          notebook_task:
+            notebook_path: ./notebooks/04_system_table_analysis.py
+            base_parameters:
+              environment: ${var.environment}
+
+        - task_key: job_cleanup
+          depends_on:
+            - task_key: system_table_analysis
+          notebook_task:
+            notebook_path: ./notebooks/01_api_job_cleanup.py
+            base_parameters:
+              environment: ${var.environment}
+
+        - task_key: dashboard_cleanup
+          depends_on:
+            - task_key: system_table_analysis
+          notebook_task:
+            notebook_path: ./notebooks/02_api_dashboard_cleanup.py
+            base_parameters:
+              environment: ${var.environment}
+
+        - task_key: vector_cleanup
+          depends_on:
+            - task_key: system_table_analysis
+          notebook_task:
+            notebook_path: ./notebooks/03_api_vector_cleanup.py
+            base_parameters:
+              environment: ${var.environment}
+
+        - task_key: system_table_cleanup
+          depends_on:
+            - task_key: job_cleanup
+            - task_key: dashboard_cleanup
+            - task_key: vector_cleanup
+          notebook_task:
+            notebook_path: ./notebooks/05_system_table_cleanup.py
+            base_parameters:
+              environment: ${var.environment}
+
+targets:
+  dev:
+    default: true
+    variables:
+      environment: dev
+  stage:
+    variables:
+      environment: stage
+  prod:
+    variables:
+      environment: prod
diff --git a/2026-04-workspace-maintenance-cleanup/notebooks/00_cleanup_logger.py b/2026-04-workspace-maintenance-cleanup/notebooks/00_cleanup_logger.py
new file mode 100644
index 0000000..d2ed19f
--- /dev/null
+++ b/2026-04-workspace-maintenance-cleanup/notebooks/00_cleanup_logger.py
@@ -0,0 +1,56 @@
+# Databricks notebook source
+# Structured logging for all cleanup operations — writes to Delta table
+
+import json
+from datetime import datetime
+
+
+class CleanupLogger:
+    """Log every cleanup action to a Delta table for auditability."""
+
+    def __init__(self, spark, catalog="finops", schema="cleanup"):
+        self.spark = spark
+        self.table = f"{catalog}.{schema}.cleanup_log"
+        self.entries = []
+        self._ensure_table(catalog, schema)
+
+    def _ensure_table(self, catalog, schema):
+        self.spark.sql(f"CREATE SCHEMA IF NOT EXISTS {catalog}.{schema}")
+        self.spark.sql(f"""
+            CREATE TABLE IF NOT EXISTS {self.table} (
+                timestamp TIMESTAMP,
+                environment STRING,
+                resource_type STRING,
+                resource_id STRING,
+                resource_name STRING,
+                owner STRING,
+                action STRING,
+                reason STRING,
+                dry_run BOOLEAN,
+                details STRING
+            )
+        """)
+
+    def log(self, environment, resource_type, resource_id, resource_name,
+            owner, action, reason, dry_run=False, details=None):
+        self.entries.append({
+            "timestamp": datetime.utcnow(),
+            "environment": environment,
+            "resource_type": resource_type,
+            "resource_id": str(resource_id),
+            "resource_name": resource_name,
+            "owner": owner or "unknown",
+            "action": action,
+            "reason": reason,
+            "dry_run": dry_run,
+            "details": json.dumps(details) if details else None
+        })
+
+    def flush(self):
+        if not self.entries:
+            return 0
+        df = self.spark.createDataFrame(self.entries)
+        df.write.mode("append").saveAsTable(self.table)
+        count = len(self.entries)
+        self.entries = []
+        return count
diff --git a/2026-04-workspace-maintenance-cleanup/notebooks/01_api_job_cleanup.py b/2026-04-workspace-maintenance-cleanup/notebooks/01_api_job_cleanup.py
new file mode 100644
index 0000000..2614bbe
--- /dev/null
+++ b/2026-04-workspace-maintenance-cleanup/notebooks/01_api_job_cleanup.py
@@ -0,0 +1,110 @@
+# Databricks notebook source
+# API-Driven Job Cleanup — deletes jobs inactive beyond threshold
+
+import requests
+import yaml
+from datetime import datetime, timedelta
+
+# COMMAND ----------
+
+dbutils.widgets.text("environment", "dev")
+env = dbutils.widgets.get("environment")
+
+with open("/Workspace/config/config.yaml") as f:
+    config = yaml.safe_load(f)[env]
+with open("/Workspace/config/thresholds.yaml") as f:
+    thresholds = yaml.safe_load(f)
+
+if not config.get("job_cleanup", False):
+    dbutils.notebook.exit(f"Job cleanup disabled for {env}")
+
+dry_run = config.get("dry_run", True)
+inactive_days = thresholds.get("job_inactive_days", 90)
+cutoff = datetime.utcnow() - timedelta(days=inactive_days)
+
+# COMMAND ----------
+
+# Setup
+%run ./00_cleanup_logger
+
+host = spark.conf.get("spark.databricks.workspaceUrl")
+token = (dbutils.notebook.entry_point.getDbutils()
+         .notebook().getContext().apiToken().get())
+headers = {"Authorization": f"Bearer {token}"}
+logger = CleanupLogger(spark)
+
+# COMMAND ----------
+
+# List all jobs
+has_more = True
+offset = 0
+jobs = []
+
+while has_more:
+    resp = requests.get(
+        f"https://{host}/api/2.1/jobs/list",
+        headers=headers,
+        params={"limit": 25, "offset": offset, "expand_tasks": False}
+    )
+    data = resp.json()
+    jobs.extend(data.get("jobs", []))
+    has_more = data.get("has_more", False)
+    offset += 25
+
+print(f"Found {len(jobs)} jobs in workspace")
+
+# COMMAND ----------
+
+deleted, skipped = 0, 0
+
+for job in jobs:
+    job_id = job["job_id"]
+    job_name = job.get("settings", {}).get("name", "unnamed")
+    creator = job.get("creator_user_name", "unknown")
+
+    # Get last run
+    runs_resp = requests.get(
+        f"https://{host}/api/2.1/jobs/runs/list",
+        headers=headers,
+        params={"job_id": job_id, "limit": 1}
+    )
+    runs = runs_resp.json().get("runs", [])
+
+    if runs:
+        last_run = datetime.utcfromtimestamp(
+            runs[0].get("start_time", 0) / 1000
+        )
+    else:
+        last_run = datetime(2020, 1, 1)
+
+    days_idle = (datetime.utcnow() - last_run).days
+
+    if last_run < cutoff:
+        if not dry_run:
+            requests.post(
+                f"https://{host}/api/2.1/jobs/delete",
+                headers=headers, json={"job_id": job_id}
+            )
+
+        logger.log(
+            environment=env, resource_type="job",
+            resource_id=job_id, resource_name=job_name, owner=creator,
+            action="DELETED" if not dry_run else "FLAGGED",
+            reason=f"Inactive {days_idle} days (threshold: {inactive_days})",
+            dry_run=dry_run,
+            details={"last_run": str(last_run), "total_runs": len(runs)}
+        )
+        deleted += 1
+    else:
+        logger.log(
+            environment=env, resource_type="job",
+            resource_id=job_id, resource_name=job_name, owner=creator,
+            action="SKIPPED",
+            reason=f"Active — last run {days_idle} days ago",
+            dry_run=dry_run
+        )
+        skipped += 1
+
+flushed = logger.flush()
+print(f"Jobs — {'[DRY RUN] ' if dry_run else ''}Deleted: {deleted}, "
+      f"Skipped: {skipped}, Logged: {flushed}")
diff --git a/2026-04-workspace-maintenance-cleanup/notebooks/02_api_dashboard_cleanup.py b/2026-04-workspace-maintenance-cleanup/notebooks/02_api_dashboard_cleanup.py
new file mode 100644
index 0000000..75ac664
--- /dev/null
+++ b/2026-04-workspace-maintenance-cleanup/notebooks/02_api_dashboard_cleanup.py
@@ -0,0 +1,93 @@
+# Databricks notebook source
+# API-Driven Dashboard Cleanup — removes stale or trashed dashboards
+
+import requests
+import yaml
+from datetime import datetime, timedelta
+
+# COMMAND ----------
+
+dbutils.widgets.text("environment", "dev")
+env = dbutils.widgets.get("environment")
+
+with open("/Workspace/config/config.yaml") as f:
+    config = yaml.safe_load(f)[env]
+with open("/Workspace/config/thresholds.yaml") as f:
+    thresholds = yaml.safe_load(f)
+
+if not config.get("dashboard_cleanup", False):
+    dbutils.notebook.exit(f"Dashboard cleanup disabled for {env}")
+
+dry_run = config.get("dry_run", True)
+inactive_days = thresholds.get("dashboard_inactive_days", 60)
+cutoff = datetime.utcnow() - timedelta(days=inactive_days)
+
+# COMMAND ----------
+
+%run ./00_cleanup_logger
+
+host = spark.conf.get("spark.databricks.workspaceUrl")
+token = (dbutils.notebook.entry_point.getDbutils()
+         .notebook().getContext().apiToken().get())
+headers = {"Authorization": f"Bearer {token}"}
+logger = CleanupLogger(spark)
+
+# COMMAND ----------
+
+resp = requests.get(
+    f"https://{host}/api/2.0/lakeview/dashboards",
+    headers=headers, params={"page_size": 100}
+)
+dashboards = resp.json().get("dashboards", [])
+print(f"Found {len(dashboards)} dashboards")
+
+# COMMAND ----------
+
+deleted, skipped = 0, 0
+
+for dash in dashboards:
+    dash_id = dash["dashboard_id"]
+    dash_name = dash.get("display_name", "unnamed")
+    creator = dash.get("creator_user_name", "unknown")
+    update_time = dash.get("update_time", "")
+    lifecycle = dash.get("lifecycle_state", "ACTIVE")
+
+    if update_time:
+        last_updated = datetime.fromisoformat(
+            update_time.replace("Z", "+00:00")
+        ).replace(tzinfo=None)
+    else:
+        last_updated = datetime(2020, 1, 1)
+
+    days_stale = (datetime.utcnow() - last_updated).days
+
+    if last_updated < cutoff or lifecycle == "TRASHED":
+        if not dry_run:
+            requests.delete(
+                f"https://{host}/api/2.0/lakeview/dashboards/{dash_id}",
+                headers=headers
+            )
+
+        logger.log(
+            environment=env, resource_type="dashboard",
+            resource_id=dash_id, resource_name=dash_name, owner=creator,
+            action="DELETED" if not dry_run else "FLAGGED",
+            reason=f"Stale {days_stale} days" if lifecycle != "TRASHED"
+                   else "Already trashed",
+            dry_run=dry_run,
+            details={"last_updated": str(last_updated), "state": lifecycle}
+        )
+        deleted += 1
+    else:
+        logger.log(
+            environment=env, resource_type="dashboard",
+            resource_id=dash_id, resource_name=dash_name, owner=creator,
+            action="SKIPPED",
+            reason=f"Active — updated {days_stale} days ago",
+            dry_run=dry_run
+        )
+        skipped += 1
+
+flushed = logger.flush()
+print(f"Dashboards — {'[DRY RUN] ' if dry_run else ''}Deleted: {deleted}, "
+      f"Skipped: {skipped}, Logged: {flushed}")
diff --git a/2026-04-workspace-maintenance-cleanup/notebooks/03_api_vector_cleanup.py b/2026-04-workspace-maintenance-cleanup/notebooks/03_api_vector_cleanup.py
new file mode 100644
index 0000000..ad9cf9f
--- /dev/null
+++ b/2026-04-workspace-maintenance-cleanup/notebooks/03_api_vector_cleanup.py
@@ -0,0 +1,94 @@
+# Databricks notebook source
+# API-Driven Vector Search Cleanup — removes orphaned indexes
+
+import requests
+import yaml
+from cleanup_logger import CleanupLogger
+
+# COMMAND ----------
+
+dbutils.widgets.text("environment", "dev")
+env = dbutils.widgets.get("environment")
+
+with open("/Workspace/config/config.yaml") as f:
+    config = yaml.safe_load(f)[env]
+
+if not config.get("vector_cleanup", False):
+    dbutils.notebook.exit(f"Vector cleanup disabled for {env}")
+
+dry_run = config.get("dry_run", True)
+
+# COMMAND ----------
+
+%run ./00_cleanup_logger
+
+host = spark.conf.get("spark.databricks.workspaceUrl")
+token = (dbutils.notebook.entry_point.getDbutils()
+         .notebook().getContext().apiToken().get())
+headers = {"Authorization": f"Bearer {token}"}
+logger = CleanupLogger(spark)
+
+# COMMAND ----------
+
+ep_resp = requests.get(
+    f"https://{host}/api/2.0/vector-search/endpoints", headers=headers
+)
+endpoints = ep_resp.json().get("endpoints", [])
+print(f"Found {len(endpoints)} vector search endpoints")
+
+# COMMAND ----------
+
+deleted, skipped = 0, 0
+
+for ep in endpoints:
+    ep_name = ep["name"]
+
+    idx_resp = requests.get(
+        f"https://{host}/api/2.0/vector-search/indexes",
+        headers=headers, params={"endpoint_name": ep_name}
+    )
+    indexes = idx_resp.json().get("vector_indexes", [])
+
+    for idx in indexes:
+        idx_name = idx["name"]
+        idx_ready = idx.get("status", {}).get("ready", False)
+        creator = idx.get("creator", "unknown")
+        source_table = idx.get("primary_key", {}).get("source_table", "")
+
+        table_exists = True
+        if source_table:
+            try:
+                spark.sql(f"DESCRIBE TABLE {source_table}")
+            except Exception:
+                table_exists = False
+
+        if not table_exists or not idx_ready:
+            if not dry_run:
+                requests.delete(
+                    f"https://{host}/api/2.0/vector-search/indexes/{idx_name}",
+                    headers=headers
+                )
+
+            logger.log(
+                environment=env, resource_type="vector_index",
+                resource_id=idx_name, resource_name=idx_name, owner=creator,
+                action="DELETED" if not dry_run else "FLAGGED",
+                reason="Source table missing" if not table_exists
+                       else "Index not ready",
+                dry_run=dry_run,
+                details={"endpoint": ep_name, "source_table": source_table}
+            )
+            deleted += 1
+        else:
+            logger.log(
+                environment=env, resource_type="vector_index",
+                resource_id=idx_name, resource_name=idx_name, owner=creator,
+                action="SKIPPED",
+                reason="Active — source exists, index ready",
+                dry_run=dry_run
+            )
+            skipped += 1
+
+flushed = logger.flush()
+print(f"Vector Indexes — {'[DRY RUN] ' if dry_run else ''}Deleted: {deleted}, "
+      f"Skipped: {skipped}, Logged: {flushed}")
diff --git a/2026-04-workspace-maintenance-cleanup/notebooks/04_system_table_analysis.py b/2026-04-workspace-maintenance-cleanup/notebooks/04_system_table_analysis.py
new file mode 100644
index 0000000..447e2ba
--- /dev/null
+++ b/2026-04-workspace-maintenance-cleanup/notebooks/04_system_table_analysis.py
@@ -0,0 +1,261 @@
+# Databricks notebook source
+# System Table-Driven Analysis — discover waste using billing + activity data
+
+# COMMAND ----------
+
+dbutils.widgets.text("environment", "dev")
+env = dbutils.widgets.get("environment")
+
+# COMMAND ----------
+
+%run ./00_cleanup_logger
+logger = CleanupLogger(spark)
+
+# COMMAND ----------
+
+# MAGIC %md
+# MAGIC ## 1. Jobs: Cost vs Activity
+
+# COMMAND ----------
+
+idle_jobs = spark.sql("""
+    WITH job_activity AS (
+        SELECT
+            job_id,
+            MAX(period_start_time) AS last_run,
+            COUNT(*) AS total_runs,
+            SUM(CASE WHEN result_state NOT IN ('SUCCESS','SUCCEEDED')
+                THEN 1 ELSE 0 END) AS failed_runs,
+            SUM(run_duration_seconds) / 3600.0 AS total_hours
+        FROM system.lakeflow.job_run_timeline
+        WHERE period_start_time >= DATEADD(DAY, -180, CURRENT_DATE())
+        GROUP BY job_id
+    ),
+    job_costs AS (
+        SELECT
+            usage_metadata.job_id AS job_id,
+            ROUND(SUM(usage_quantity * p.pricing.default), 2) AS cost_90d,
+            SUM(usage_quantity) AS dbus_90d
+        FROM system.billing.usage u
+        JOIN system.billing.list_prices p
+            ON u.sku_name = p.sku_name AND u.cloud = p.cloud
+            AND u.usage_start_time >= p.price_start_time
+            AND (p.price_end_time IS NULL
+                 OR u.usage_start_time < p.price_end_time)
+        WHERE u.usage_date >= DATEADD(DAY, -90, CURRENT_DATE())
+            AND u.usage_metadata.job_id IS NOT NULL
+        GROUP BY usage_metadata.job_id
+    )
+    SELECT
+        ja.job_id,
+        ja.last_run,
+        DATEDIFF(DAY, ja.last_run, CURRENT_TIMESTAMP()) AS days_idle,
+        ja.total_runs,
+        ja.failed_runs,
+        ROUND(ja.total_hours, 2) AS total_hours,
+        COALESCE(jc.cost_90d, 0) AS cost_90d,
+        COALESCE(jc.dbus_90d, 0) AS dbus_90d,
+        CASE
+            WHEN DATEDIFF(DAY, ja.last_run, CURRENT_TIMESTAMP()) > 90
+                THEN 'CANDIDATE_DELETE'
+            WHEN ja.failed_runs > ja.total_runs * 0.8
+                THEN 'CANDIDATE_REVIEW'
+            WHEN COALESCE(jc.cost_90d, 0) > 1000 AND ja.total_runs < 5
+                THEN 'CANDIDATE_REVIEW'
+            ELSE 'HEALTHY'
+        END AS recommendation
+    FROM job_activity ja
+    LEFT JOIN job_costs jc ON ja.job_id = jc.job_id
+    ORDER BY cost_90d DESC
+""")
+
+idle_jobs.createOrReplaceTempView("job_analysis")
+display(idle_jobs)
+
+# COMMAND ----------
+
+# MAGIC %md
+# MAGIC ## 2. Clusters: Cost vs Utilisation
+
+# COMMAND ----------
+
+cluster_analysis = spark.sql("""
+    WITH cluster_costs AS (
+        SELECT
+            usage_metadata.cluster_id AS cluster_id,
+            ROUND(SUM(usage_quantity * p.pricing.default), 2) AS cost_30d,
+            SUM(usage_quantity) AS dbus_30d,
+            COUNT(DISTINCT usage_date) AS active_days
+        FROM system.billing.usage u
+        JOIN system.billing.list_prices p
+            ON u.sku_name = p.sku_name AND u.cloud = p.cloud
+            AND u.usage_start_time >= p.price_start_time
+            AND (p.price_end_time IS NULL
+                 OR u.usage_start_time < p.price_end_time)
+        WHERE u.usage_date >= DATEADD(DAY, -30, CURRENT_DATE())
+            AND u.usage_metadata.cluster_id IS NOT NULL
+        GROUP BY usage_metadata.cluster_id
+    ),
+    cluster_info AS (
+        SELECT cluster_id, cluster_name, owned_by AS owner,
+               cluster_source, driver_node_type, worker_node_type
+        FROM system.compute.clusters
+        WHERE delete_time IS NULL
+    )
+    SELECT
+        ci.cluster_id, ci.cluster_name, ci.owner,
+        ci.cluster_source,
+        COALESCE(cc.cost_30d, 0) AS cost_30d,
+        COALESCE(cc.dbus_30d, 0) AS dbus_30d,
+        COALESCE(cc.active_days, 0) AS active_days,
+        CASE
+            WHEN COALESCE(cc.active_days, 0) = 0 THEN 'CANDIDATE_DELETE'
+            WHEN cc.cost_30d > 5000 AND cc.active_days < 5
+                THEN 'CANDIDATE_REVIEW'
+            WHEN ci.cluster_source = 'UI' THEN 'CANDIDATE_REVIEW'
+            ELSE 'HEALTHY'
+        END AS recommendation
+    FROM cluster_info ci
+    LEFT JOIN cluster_costs cc ON ci.cluster_id = cc.cluster_id
+    ORDER BY cost_30d DESC
+""")
+
+cluster_analysis.createOrReplaceTempView("cluster_analysis")
+display(cluster_analysis)
+
+# COMMAND ----------
+
+# MAGIC %md
+# MAGIC ## 3. SQL Warehouses: Cost vs Query Volume
+
+# COMMAND ----------
+
+warehouse_analysis = spark.sql("""
+    WITH wh_costs AS (
+        SELECT
+            usage_metadata.warehouse_id AS warehouse_id,
+            ROUND(SUM(usage_quantity * p.pricing.default), 2) AS cost_30d,
+            COUNT(DISTINCT usage_date) AS billing_days
+        FROM system.billing.usage u
+        JOIN system.billing.list_prices p
+            ON u.sku_name = p.sku_name AND u.cloud = p.cloud
+            AND u.usage_start_time >= p.price_start_time
+            AND (p.price_end_time IS NULL
+                 OR u.usage_start_time < p.price_end_time)
+        WHERE u.usage_date >= DATEADD(DAY, -30, CURRENT_DATE())
+            AND u.usage_metadata.warehouse_id IS NOT NULL
+            AND u.sku_name LIKE '%SQL%'
+        GROUP BY usage_metadata.warehouse_id
+    ),
+    wh_queries AS (
+        SELECT compute.warehouse_id AS warehouse_id,
+               COUNT(*) AS queries_30d,
+               COUNT(DISTINCT DATE(start_time)) AS query_days
+        FROM system.query.history
+        WHERE start_time >= DATEADD(DAY, -30, CURRENT_DATE())
+            AND compute.warehouse_id IS NOT NULL
+        GROUP BY compute.warehouse_id
+    )
+    SELECT
+        wc.warehouse_id, wc.cost_30d, wc.billing_days,
+        COALESCE(wq.queries_30d, 0) AS queries_30d,
+        COALESCE(wq.query_days, 0) AS query_days,
+        CASE
+            WHEN COALESCE(wq.queries_30d, 0) = 0 THEN 'CANDIDATE_DELETE'
+            WHEN wc.cost_30d > 1000 AND wq.queries_30d < 10
+                THEN 'CANDIDATE_REVIEW'
+            ELSE 'HEALTHY'
+        END AS recommendation
+    FROM wh_costs wc
+    LEFT JOIN wh_queries wq ON wc.warehouse_id = wq.warehouse_id
+    ORDER BY cost_30d DESC
+""")
+
+warehouse_analysis.createOrReplaceTempView("warehouse_analysis")
+display(warehouse_analysis)
+
+# COMMAND ----------
+
+# MAGIC %md
+# MAGIC ## 4. Model Serving Endpoints: Cost vs Traffic
+
+# COMMAND ----------
+
+serving_analysis = spark.sql("""
+    WITH ep_costs AS (
+        SELECT
+            usage_metadata.endpoint_id AS endpoint_id,
+            usage_metadata.endpoint_name AS endpoint_name,
+            ROUND(SUM(usage_quantity * p.pricing.default), 2) AS cost_30d
+        FROM system.billing.usage u
+        JOIN system.billing.list_prices p
+            ON u.sku_name = p.sku_name AND u.cloud = p.cloud
+            AND u.usage_start_time >= p.price_start_time
+            AND (p.price_end_time IS NULL
+                 OR u.usage_start_time < p.price_end_time)
+        WHERE u.usage_date >= DATEADD(DAY, -30, CURRENT_DATE())
+            AND u.sku_name LIKE '%SERVING%'
+            AND u.usage_metadata.endpoint_id IS NOT NULL
+        GROUP BY usage_metadata.endpoint_id, usage_metadata.endpoint_name
+    ),
+    ep_traffic AS (
+        SELECT served_entity_id,
+               COUNT(*) AS requests_30d
+        FROM system.serving.endpoint_usage
+        WHERE request_time >= DATEADD(DAY, -30, CURRENT_DATE())
+        GROUP BY served_entity_id
+    )
+    SELECT
+        ec.endpoint_id, ec.endpoint_name, ec.cost_30d,
+        COALESCE(SUM(et.requests_30d), 0) AS requests_30d,
+        CASE
+            WHEN COALESCE(SUM(et.requests_30d), 0) = 0
+                THEN 'CANDIDATE_DELETE'
+            WHEN ec.cost_30d > 500
+                 AND COALESCE(SUM(et.requests_30d), 0) < 100
+                THEN 'CANDIDATE_REVIEW'
+            ELSE 'HEALTHY'
+        END AS recommendation
+    FROM ep_costs ec
+    LEFT JOIN system.serving.served_entities se
+        ON ec.endpoint_id = se.endpoint_id
+    LEFT JOIN ep_traffic et
+        ON se.served_entity_id = et.served_entity_id
+    GROUP BY ec.endpoint_id, ec.endpoint_name, ec.cost_30d
+    ORDER BY cost_30d DESC
+""")
+
+serving_analysis.createOrReplaceTempView("serving_analysis")
+display(serving_analysis)
+
+# COMMAND ----------
+
+# MAGIC %md
+# MAGIC ## Log All Flagged Items
+
+# COMMAND ----------
+
+for view_name, res_type in [("job_analysis", "job"), ("cluster_analysis", "cluster"),
+                              ("warehouse_analysis", "sql_warehouse"),
+                              ("serving_analysis", "serving_endpoint")]:
+    try:
+        flagged = spark.sql(f"SELECT * FROM {view_name} WHERE recommendation != 'HEALTHY'").collect()
+        for row in flagged:
+            row_dict = row.asDict()
+            cost = row_dict.get("cost_90d", row_dict.get("cost_30d", 0))
+            logger.log(
+                environment=env, resource_type=res_type,
+                resource_id=str(row_dict.get("job_id", row_dict.get("cluster_id",
+                    row_dict.get("warehouse_id", row_dict.get("endpoint_id", "unknown"))))),
+                resource_name=row_dict.get("cluster_name", row_dict.get("endpoint_name", f"{res_type}")),
+                owner=row_dict.get("owner", "system_scan"),
+                action="FLAGGED",
+                reason=f"{row_dict.get('recommendation', 'REVIEW')}: ${cost} cost",
+                dry_run=True,
+                details={"cost": float(cost)}
+            )
+    except Exception as e:
+        print(f"Skipping {view_name}: {e}")
+
+flushed = logger.flush()
+print(f"System table analysis complete. {flushed} items flagged.")
diff --git a/2026-04-workspace-maintenance-cleanup/notebooks/05_system_table_cleanup.py b/2026-04-workspace-maintenance-cleanup/notebooks/05_system_table_cleanup.py
new file mode 100644
index 0000000..4c1c582
--- /dev/null
+++ b/2026-04-workspace-maintenance-cleanup/notebooks/05_system_table_cleanup.py
@@ -0,0 +1,104 @@
+# Databricks notebook source
+# System Table-Driven Cleanup — acts on flagged items from analysis step
+
+import requests
+import yaml
+
+# COMMAND ----------
+
+dbutils.widgets.text("environment", "dev")
+env = dbutils.widgets.get("environment")
+
+with open("/Workspace/config/config.yaml") as f:
+    config = yaml.safe_load(f)[env]
+
+if not config.get("system_table_cleanup", False):
+    dbutils.notebook.exit(f"System table cleanup disabled for {env}")
+
+dry_run = config.get("dry_run", True)
+
+# COMMAND ----------
+
+%run ./00_cleanup_logger
+
+host = spark.conf.get("spark.databricks.workspaceUrl")
+token = (dbutils.notebook.entry_point.getDbutils()
+         .notebook().getContext().apiToken().get())
+headers = {"Authorization": f"Bearer {token}"}
+logger = CleanupLogger(spark)
+
+# COMMAND ----------
+
+# MAGIC %md
+# MAGIC ## Delete Flagged Jobs
+
+# COMMAND ----------
+
+try:
+    flagged_jobs = spark.sql("""
+        SELECT job_id, recommendation, cost_90d, days_idle
+        FROM job_analysis WHERE recommendation = 'CANDIDATE_DELETE'
+    """).collect()
+except Exception:
+    flagged_jobs = []
+
+print(f"{'[DRY RUN] ' if dry_run else ''}{len(flagged_jobs)} jobs to delete")
+
+for row in flagged_jobs:
+    if not dry_run:
+        resp = requests.post(
+            f"https://{host}/api/2.1/jobs/delete",
+            headers=headers, json={"job_id": row.job_id}
+        )
+        action = "DELETED" if resp.status_code == 200 else "FAILED"
+    else:
+        action = "DRY_RUN"
+
+    logger.log(
+        environment=env, resource_type="job",
+        resource_id=row.job_id, resource_name=f"job-{row.job_id}",
+        owner="system_table_cleanup", action=action,
+        reason=f"{row.days_idle} days idle, ${row.cost_90d} wasted",
+        dry_run=dry_run
+    )
+
+# COMMAND ----------
+
+# MAGIC %md
+# MAGIC ## Terminate Flagged Clusters
+
+# COMMAND ----------
+
+try:
+    flagged_clusters = spark.sql("""
+        SELECT cluster_id, cluster_name, recommendation, cost_30d
+        FROM cluster_analysis WHERE recommendation = 'CANDIDATE_DELETE'
+    """).collect()
+except Exception:
+    flagged_clusters = []
+
+print(f"{'[DRY RUN] ' if dry_run else ''}{len(flagged_clusters)} clusters to terminate")
+
+for row in flagged_clusters:
+    if not dry_run:
+        resp = requests.post(
+            f"https://{host}/api/2.0/clusters/permanent-delete",
+            headers=headers, json={"cluster_id": row.cluster_id}
+        )
+        action = "DELETED" if resp.status_code == 200 else "FAILED"
+    else:
+        action = "DRY_RUN"
+
+    logger.log(
+        environment=env, resource_type="cluster",
+        resource_id=row.cluster_id,
+        resource_name=row.cluster_name or "unnamed",
+        owner="system_table_cleanup", action=action,
+        reason=f"Zero activity, ${row.cost_30d} wasted in 30d",
+        dry_run=dry_run
+    )
+
+# COMMAND ----------
+
+flushed = logger.flush()
+print(f"\nCleanup complete. {flushed} actions logged. Mode: {'DRY RUN' if dry_run else 'LIVE'}")