feat: Spark Workbench everywhere, autonomous Hadoop offload & LLM masking-aware

- Data Hub with Hadoop tab (HDFS/Iceberg browser, Spark, pipeline)
- Databricks-style Lakehouse Workbench (Trino engine, live exec matrix,
  materialize to Iceberg/S3); reused & embedded in every source-DB UI
- HDFS -> Kafka -> Spark -> Iceberg/S3 pipeline; WebHDFS hostname resolver
- Data Flow master pulse switch (Run/Pause/Stop) gating animated edges
- Data Custodian autonomous Hadoop offload loop (batch counterpart to CDC),
  pulsing source -> HDFS edges; toggle in Data Flow
- LLM now autonomously aware of all latest platform changes (live platform
  context) and enforces masking policy: never reveals masked PII, still
  answers helpfully with aggregates/explanations

api/agent_ops.py

@@ -328,11 +328,98 @@ async def etl_agent_loop() -> None:
         await asyncio.sleep(max(30.0, float(_etl_state["interval"])))
 
 
+
+# ── Custodian Hadoop offload (batch counterpart to CDC) ─────────────────────
+_CUST_INTERVAL = float(os.getenv("CUSTODIAN_OFFLOAD_INTERVAL_SECONDS", "120"))
+_CUST_BATCH = int(os.getenv("CUSTODIAN_OFFLOAD_BATCH", "200"))
+_CUST_TARGETS = [
+    {"label": "postgres sales_orders", "src": "postgres_sales.public.sales_orders",
+     "target": "iceberg.hadoop.sales_orders_offload"},
+    {"label": "mysql employee_events", "src": "mysql_hr.hr.employee_events",
+     "target": "iceberg.hadoop.employee_events_offload"},
+]
+_custodian_state: dict[str, Any] = {
+    "enabled": os.getenv("CUSTODIAN_OFFLOAD_ENABLED", "1") not in ("0", "false", "False", ""),
+    "interval": _CUST_INTERVAL,
+    "targets": [c["target"] for c in _CUST_TARGETS],
+    "idx": 0,
+    "runs_total": 0,
+    "last": None,
+    "started": False,
+}
+
+
+async def _custodian_offload_once(idx: int | None = None) -> dict[str, Any]:
+    """Offload a batch of source rows into the Hadoop Iceberg lake via Trino."""
+    from spark_workbench import _trino_collect
+    i = _custodian_state["idx"] if idx is None else idx
+    tgt = _CUST_TARGETS[i % len(_CUST_TARGETS)]
+    _custodian_state["idx"] = i + 1
+    await _trino_collect(
+        f"CREATE TABLE IF NOT EXISTS {tgt['target']} AS SELECT * FROM {tgt['src']} WHERE 1=0", 1)
+    ins = await _trino_collect(
+        f"INSERT INTO {tgt['target']} SELECT * FROM {tgt['src']} LIMIT {_CUST_BATCH}", 1)
+    ok = bool(ins.get("ok"))
+    _custodian_state["runs_total"] += 1
+    _custodian_state["last"] = {
+        "target": tgt["target"], "src": tgt["src"], "ok": ok,
+        "rows": _CUST_BATCH if ok else 0,
+        "ts": datetime.now(timezone.utc).isoformat(), "error": ins.get("error"),
+    }
+    if ok:
+        await _emit(f"[custodian-offload] {tgt['label']} → {tgt['target']}: offloaded ~{_CUST_BATCH} rows to Hadoop", "info")
+    else:
+        await _emit(f"[custodian-offload] {tgt['label']} failed: {str(ins.get('error'))[:120]}", "err")
+    return _custodian_state["last"]
+
+
+async def custodian_offload_loop() -> None:
+    _custodian_state["started"] = True
+    await asyncio.sleep(45)
+    await _emit("[custodian-offload] Autonomous Hadoop offload online — batching source data into the Iceberg lake", "info")
+    while True:
+        try:
+            if _custodian_state["enabled"]:
+                await _custodian_offload_once()
+        except Exception as exc:
+            _custodian_state["last"] = {"error": str(exc), "ts": datetime.now(timezone.utc).isoformat()}
+        await asyncio.sleep(max(30.0, float(_custodian_state["interval"])))
+
+
+def custodian_recent() -> bool:
+    last = _custodian_state.get("last") or {}
+    ts = last.get("ts")
+    if not ts or not last.get("ok"):
+        return False
+    try:
+        from datetime import datetime as _dt
+        t = _dt.fromisoformat(str(ts).replace("Z", "+00:00"))
+        window = max(60.0, float(_custodian_state["interval"]) * 1.5)
+        return (datetime.now(timezone.utc) - t).total_seconds() < window
+    except Exception:
+        return False
+
+
 # ── Endpoints ────────────────────────────────────────────────────────────────
 @router.get("/status")
 async def status() -> JSONResponse:
     pools = {k: len(v) for k, v in _pools.items()}
-    return JSONResponse({"ok": True, "pools": pools, "etl": _etl_state, **_state})
+    return JSONResponse({"ok": True, "pools": pools, "etl": _etl_state, "custodian": _custodian_state, **_state})
+
+
+@router.post("/custodian/toggle")
+async def custodian_toggle(body: dict[str, Any] = Body(default={})) -> JSONResponse:
+    if "enabled" in body:
+        _custodian_state["enabled"] = bool(body["enabled"])
+    else:
+        _custodian_state["enabled"] = not _custodian_state["enabled"]
+    if "interval" in body:
+        try:
+            _custodian_state["interval"] = max(30.0, float(body["interval"]))
+        except (TypeError, ValueError):
+            pass
+    await _emit(f"[custodian-offload] Hadoop offload {'ENABLED' if _custodian_state['enabled'] else 'PAUSED'} by operator", "warn")
+    return JSONResponse({"ok": True, "enabled": _custodian_state["enabled"], "interval": _custodian_state["interval"]})
 
 
 @router.post("/etl/toggle")