feat: realtime ETL offload to S3 + live business dashboards

- etl_offload.py: autonomous agent backfills/tails source DBs (PG/MySQL/
  Mongo/Cassandra) to S3 as Parquet in small chunks, accumulates a live
  federated business matrix (/api/etl/status, /api/etl/business, /run, /config).
- storage_s3.py: buffer generated CDC + masked curated rows to S3, overlay
  live last-write into analytics; put_object_bytes for Parquet parts.
- trino_federated.py: capture generated rows + archive to S3; generator_active.
- dataflow.py: pulse generate + kafka/spark->S3 archive edges when active.
- StorageView: realtime ETL ingest panel; TrinoFederationView: realtime
  business KPIs/charts from /api/etl/business.
- ChangesView: top KPIs/charts now overlay the live WS stream on server stats
  so they update in lock-step with the bottom feed; faster 2.5s refresh.
- useCommandCenter: retain 800 live CDC changes.

api/Dockerfile

@@ -4,7 +4,7 @@ WORKDIR /app
 RUN apt-get update && apt-get install -y --no-install-recommends curl && rm -rf /var/lib/apt/lists/*
 COPY requirements.txt .
 RUN pip install --no-cache-dir -r requirements.txt
-COPY main.py lab_context.py agent_terminal.py workload.py node_registry.py node_ops.py topology_views.py supervisor.py approval_service.py db.py dockhand_envs.py presentation.py database_inventory.py presentation_upload.py presentation_static.py storage_s3.py elasticsearch_api.py sql_console.py hdfs_api.py ssh_terminal.py pipeline_ops.py hadoop_analytics.py agent_ops.py cdc_consumer.py movements.py dataflow.py streaming_ops.py spark_workbench.py hadoop_sql.py hdfs_kafka.py webhdfs_util.py pii_catalog.py platform_context.py trino_federated.py hive_bench_seed.json .
+COPY main.py lab_context.py agent_terminal.py workload.py node_registry.py node_ops.py topology_views.py supervisor.py approval_service.py db.py dockhand_envs.py presentation.py database_inventory.py presentation_upload.py presentation_static.py storage_s3.py elasticsearch_api.py sql_console.py hdfs_api.py ssh_terminal.py pipeline_ops.py hadoop_analytics.py agent_ops.py cdc_consumer.py movements.py dataflow.py streaming_ops.py spark_workbench.py hadoop_sql.py hdfs_kafka.py webhdfs_util.py pii_catalog.py platform_context.py trino_federated.py etl_offload.py hive_bench_seed.json .
 RUN mkdir -p /data
 ENV DATABASE_URL=sqlite:////data/atc-agents.db
 EXPOSE 3201

api/dataflow.py

@@ -178,6 +178,11 @@ async def _build() -> dict[str, Any]:
         gen_active = generator_active()
     except Exception:
         gen_active = False
+    try:
+        from storage_s3 import archive_active
+        arch_active = archive_active()
+    except Exception:
+        arch_active = False
     try:
         from pii_catalog import get_pii
         pii = get_pii()
@@ -274,7 +279,10 @@ async def _build() -> dict[str, Any]:
         elif e.get("from") == "spark" and e.get("to") == "iceberg_curated":
             edge["active"] = bool(edge_live.get("spark→iceberg")) or edge.get("active")
         elif e.get("from") == "spark" and e.get("to") == "s3_cdc":
-            edge["active"] = bool(edge_live.get("spark→s3")) or edge.get("active")
+            edge["active"] = bool(edge_live.get("spark→s3")) or edge.get("active") or arch_active
+        elif e["kind"] == "archive" and e.get("from") == "kafka" and e.get("to") == "s3_cdc":
+            # Kafka → S3 CDC archive pulses while the pipeline lands objects in S3
+            edge["active"] = arch_active
         elif e["kind"] in ("context", "retrieve", "prompt", "answer"):
             # AI serving lane pulses while governed data is being served to the LLM
             edge["active"] = bool(_rag_info())

api/etl_offload.py

@@ -0,0 +1,556 @@
+"""Autonomous ETL offload agent — source databases → S3 Parquet lake.
+
+A background agent ("Lakehouse Loader") that, on a fixed cadence, pulls the next
+small chunk of rows from every source database and writes it to the S3 object
+store as a partitioned Parquet part:
+
+    lake/<dataset>/dt=YYYY-MM-DD/part-<ts>.parquet
+
+It progressively *backfills* the entire history in small chunks (so a 50M-row
+table lands as thousands of small files) and then *tails* newly generated rows,
+so the object store keeps filling and the analytics stay realtime. While
+offloading it also accumulates a live federated business matrix (revenue by
+region/status/channel, HR by dept, supply by type, telemetry by metric, …)
+straight from the rows it actually moved — which therefore always reflects the
+newest generated data, and powers the realtime Trino / Object-store dashboards.
+
+Endpoints:
+  GET  /api/etl/status     -> per-dataset progress, ingest rate, recent parts, feed
+  GET  /api/etl/business   -> accumulated realtime federated business matrix
+  POST /api/etl/run        -> trigger one offload cycle now
+  POST /api/etl/config     -> {enabled, interval_s, chunk}
+"""
+
+from __future__ import annotations
+
+import io
+import json
+import random
+import threading
+import time
+from collections import deque
+from datetime import datetime, timezone
+from typing import Any
+
+from fastapi import APIRouter, Body
+from fastapi.responses import JSONResponse
+
+router = APIRouter(prefix="/api/etl", tags=["etl"])
+
+DATASETS = [
+    {"key": "orders", "label": "Sales orders", "engine": "PostgreSQL", "color": "#fbbf24"},
+    {"key": "hr_events", "label": "HR events", "engine": "MySQL", "color": "#60a5fa"},
+    {"key": "supply_events", "label": "Supply events", "engine": "MongoDB", "color": "#a78bfa"},
+    {"key": "telemetry", "label": "Device telemetry", "engine": "Cassandra", "color": "#22d3ee"},
+]
+
+_lock = threading.Lock()
+_state: dict[str, Any] = {
+    "enabled": True,
+    "interval_s": 60.0,
+    "chunk": 5000,
+    "running_cycle": False,
+    "started_at": None,
+    "last_cycle_ts": 0.0,
+    "last_cycle_rows": 0,
+    "next_run_ts": 0.0,
+    "cycles": 0,
+    "datasets": {
+        d["key"]: {"label": d["label"], "engine": d["engine"], "color": d["color"],
+                   "parts": 0, "rows": 0, "bytes": 0, "backfilled": False, "cursor": None,
+                   "keycol": None, "mode": None, "total_source": None, "last_ts": None,
+                   "last_rows": 0, "last_key": None, "error": None}
+        for d in DATASETS
+    },
+    "feed": deque(maxlen=60),
+    "minute": deque(maxlen=60),      # (minute_epoch, rows, bytes)
+    "series": deque(maxlen=48),      # per-cycle points for the realtime chart
+}
+
+# Accumulated federated business matrix, built from the rows we actually offload.
+_business: dict[str, Any] = {
+    "orders": {"count": 0, "revenue": 0.0, "by_region": {}, "by_status": {}, "by_channel": {}, "by_currency": {}},
+    "hr": {"count": 0, "by_department": {}, "by_event": {}, "by_region": {}},
+    "supply": {"count": 0, "amount": 0.0, "by_type": {}, "by_region": {}, "by_source": {}},
+    "telemetry": {"count": 0, "by_metric": {}},
+    "ts": deque(maxlen=48),          # {t, orders, revenue, rows}
+}
+
+
+# ── helpers ───────────────────────────────────────────────────────────────────
+def _now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def _bump(d: dict, k: Any, n: int = 1, val: float | None = None) -> None:
+    if k is None or k == "":
+        return
+    k = str(k)
+    if val is None:
+        d[k] = d.get(k, 0) + n
+    else:
+        cur = d.get(k) or {"count": 0, "value": 0.0}
+        cur["count"] += n
+        cur["value"] += val
+        d[k] = cur
+
+
+def _feed(text: str, level: str = "info") -> None:
+    _state["feed"].appendleft({"ts": _now().isoformat(), "text": text, "level": level})
+    try:
+        from main import add_feed
+        add_feed("etl-guardian", f"[lakehouse] {text}", level)
+    except Exception:
+        pass
+
+
+def _scalar(v: Any) -> Any:
+    import datetime as _dt
+    import decimal
+    if v is None or isinstance(v, (str, int, float, bool)):
+        return v
+    if isinstance(v, decimal.Decimal):
+        return float(v)
+    if isinstance(v, (_dt.datetime, _dt.date)):
+        return v.isoformat()
+    if isinstance(v, (dict, list)):
+        return json.dumps(v, default=str)
+    return str(v)
+
+
+def _normalize(rows: list[dict]) -> list[dict]:
+    keys: list[str] = []
+    seen: set[str] = set()
+    for r in rows:
+        for k in r.keys():
+            if k not in seen:
+                seen.add(k)
+                keys.append(k)
+    return [{k: _scalar(r.get(k)) for k in keys} for r in rows]
+
+
+def _write_parquet(dataset_key: str, rows: list[dict]) -> tuple[str, int]:
+    import pyarrow as pa
+    import pyarrow.parquet as pq
+    table = pa.Table.from_pylist(_normalize(rows))
+    buf = io.BytesIO()
+    pq.write_table(table, buf, compression="snappy")
+    body = buf.getvalue()
+    now = _now()
+    key = (f"lake/{dataset_key}/dt={now.strftime('%Y-%m-%d')}/"
+           f"part-{int(now.timestamp() * 1000)}-{random.randint(1000, 9999)}.parquet")
+    from storage_s3 import put_object_bytes
+    put_object_bytes(key, body, "application/vnd.apache.parquet")
+    return key, len(body)
+
+
+# ── source readers: return (rows, new_cursor, done) ─────────────────────────────
+def _read_orders(cursor, n):
+    import psycopg2
+    import psycopg2.extras
+    import sql_console as s
+    conn = psycopg2.connect(host=s.DB_HOST, port=s.PG_PORT, user=s.PG_USER, password=s.PG_PASS,
+                            dbname=s.PG_DB, connect_timeout=8)
+    try:
+        conn.autocommit = True
+        cur = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
+        wm = int(cursor or 0)
+        cur.execute("SELECT * FROM public.sales_orders WHERE order_id > %s ORDER BY order_id LIMIT %s", (wm, n))
+        rows = [dict(r) for r in cur.fetchall()]
+        new_cursor = rows[-1].get("order_id", wm) if rows else wm
+        return rows, new_cursor, len(rows) < n
+    finally:
+        conn.close()
+
+
+def _mysql_keycol(cur) -> tuple[str, str]:
+    import sql_console as s
+    try:
+        cur.execute(
+            "SELECT column_name FROM information_schema.columns WHERE table_schema=%s AND table_name='employee_events' "
+            "AND extra LIKE '%%auto_increment%%' LIMIT 1", (s.MYSQL_DB,))
+        r = cur.fetchone()
+        if r:
+            return (list(r.values())[0] if isinstance(r, dict) else r[0]), "key"
+        cur.execute(
+            "SELECT column_name FROM information_schema.key_column_usage WHERE table_schema=%s "
+            "AND table_name='employee_events' AND constraint_name='PRIMARY' ORDER BY ordinal_position LIMIT 1",
+            (s.MYSQL_DB,))
+        r = cur.fetchone()
+        if r:
+            return (list(r.values())[0] if isinstance(r, dict) else r[0]), "key"
+    except Exception:
+        pass
+    return "", "offset"
+
+
+def _read_hr(cursor, n, st):
+    import pymysql
+    import sql_console as s
+    conn = pymysql.connect(host=s.DB_HOST, port=s.MYSQL_PORT, user=s.MYSQL_USER, password=s.MYSQL_PASS,
+                           database=s.MYSQL_DB, connect_timeout=8, cursorclass=pymysql.cursors.DictCursor)
+    try:
+        cur = conn.cursor()
+        if not st.get("keycol") and st.get("mode") is None:
+            kc, mode = _mysql_keycol(cur)
+            st["keycol"] = kc
+            st["mode"] = mode
+        if st.get("mode") == "key" and st.get("keycol"):
+            kc = st["keycol"]
+            wm = int(cursor or 0)
+            cur.execute(f"SELECT * FROM employee_events WHERE `{kc}` > %s ORDER BY `{kc}` LIMIT %s", (wm, n))
+            rows = list(cur.fetchall())
+            new_cursor = rows[-1].get(kc, wm) if rows else wm
+        else:  # offset fallback
+            off = int(cursor or 0)
+            cur.execute("SELECT * FROM employee_events LIMIT %s OFFSET %s", (n, off))
+            rows = list(cur.fetchall())
+            new_cursor = off + len(rows)
+        return rows, new_cursor, len(rows) < n
+    finally:
+        conn.close()
+
+
+def _read_supply(cursor, n):
+    from bson import ObjectId
+    import sql_console as s
+    cli = s._mongo_client()
+    try:
+        col = cli[s.MONGO_DB]["events"]
+        q = {"_id": {"$gt": ObjectId(cursor)}} if cursor else {}
+        docs = list(col.find(q).sort("_id", 1).limit(n))
+        new_cursor = str(docs[-1]["_id"]) if docs else cursor
+        for d in docs:
+            d["_id"] = str(d["_id"])
+        return docs, new_cursor, len(docs) < n
+    finally:
+        cli.close()
+
+
+def _read_telemetry(cursor, n):
+    from cassandra.query import SimpleStatement
+    import sql_console as s
+    cluster = s._cass_cluster()
+    try:
+        sess = cluster.connect()
+        stmt = SimpleStatement(f"SELECT * FROM {s.CASS_KS}.device_metrics", fetch_size=n)
+        kwargs = {}
+        if cursor:
+            try:
+                kwargs["paging_state"] = bytes.fromhex(cursor)
+            except Exception:
+                kwargs = {}
+        rs = sess.execute(stmt, **kwargs)
+        rows = [dict(r._asdict()) for r in rs.current_rows]
+        ps = rs.paging_state
+        new_cursor = ps.hex() if ps else None
+        return rows, new_cursor, new_cursor is None
+    finally:
+        cluster.shutdown()
+
+
+# ── business matrix accumulation ────────────────────────────────────────────────
+def _num(v) -> float:
+    try:
+        return float(v)
+    except Exception:
+        return 0.0
+
+
+def _agg(dataset_key: str, rows: list[dict]) -> tuple[int, float]:
+    """Fold a freshly-offloaded chunk into the live business matrix. Returns
+    (order_rows, revenue) for the realtime time-series."""
+    o_rows = 0
+    o_rev = 0.0
+    with _lock:
+        if dataset_key == "orders":
+            b = _business["orders"]
+            for r in rows:
+                amt = _num(r.get("amount") if r.get("amount") is not None else r.get("total_amount"))
+                b["count"] += 1
+                b["revenue"] += amt
+                _bump(b["by_region"], r.get("region"), val=amt)
+                _bump(b["by_status"], r.get("order_status") or r.get("status"))
+                _bump(b["by_channel"], r.get("sales_channel") or r.get("channel"), val=amt)
+                _bump(b["by_currency"], r.get("currency"))
+                o_rows += 1
+                o_rev += amt
+        elif dataset_key == "hr_events":
+            b = _business["hr"]
+            for r in rows:
+                b["count"] += 1
+                _bump(b["by_department"], r.get("department"))
+                _bump(b["by_event"], r.get("event_type") or r.get("event"))
+                _bump(b["by_region"], r.get("region"))
+        elif dataset_key == "supply_events":
+            b = _business["supply"]
+            for r in rows:
+                amt = _num(r.get("amount"))
+                b["count"] += 1
+                b["amount"] += amt
+                _bump(b["by_type"], r.get("type"), val=amt)
+                _bump(b["by_region"], r.get("region"))
+                _bump(b["by_source"], r.get("source"))
+        elif dataset_key == "telemetry":
+            b = _business["telemetry"]
+            for r in rows:
+                b["count"] += 1
+                mt = r.get("metric_type") or r.get("metric_name")
+                if mt:
+                    cur = b["by_metric"].get(str(mt)) or {"count": 0, "sum": 0.0}
+                    cur["count"] += 1
+                    cur["sum"] += _num(r.get("metric_value"))
+                    b["by_metric"][str(mt)] = cur
+    return o_rows, o_rev
+
+
+_READERS = {"orders": _read_orders, "hr_events": _read_hr,
+            "supply_events": _read_supply, "telemetry": _read_telemetry}
+
+
+def _offload_dataset(key: str) -> dict[str, Any]:
+    st = _state["datasets"][key]
+    chunk = int(_state["chunk"])
+    res = {"rows": 0, "bytes": 0, "order_rows": 0, "revenue": 0.0}
+    try:
+        if key == "hr_events":
+            rows, new_cursor, done = _read_hr(st["cursor"], chunk, st)
+        else:
+            rows, new_cursor, done = _READERS[key](st["cursor"], chunk)
+    except Exception as exc:
+        st["error"] = str(exc)[:160]
+        return res
+    st["error"] = None
+    if not rows:
+        if done:
+            st["backfilled"] = True
+            if key == "telemetry":  # no global order — loop back to keep tailing
+                st["cursor"] = None
+        return res
+    try:
+        obj_key, nbytes = _write_parquet(key, rows)
+    except Exception as exc:
+        st["error"] = f"parquet/s3: {str(exc)[:140]}"
+        return res
+    o_rows, o_rev = _agg(key, rows)
+    with _lock:
+        st["parts"] += 1
+        st["rows"] += len(rows)
+        st["bytes"] += nbytes
+        st["cursor"] = new_cursor
+        st["backfilled"] = bool(done)
+        st["last_ts"] = _now().isoformat()
+        st["last_rows"] = len(rows)
+        st["last_key"] = obj_key
+        if key == "telemetry" and done:
+            st["cursor"] = None
+    res.update({"rows": len(rows), "bytes": nbytes, "order_rows": o_rows, "revenue": o_rev})
+    return res
+
+
+def _source_totals() -> None:
+    """Best-effort backfill targets so the dashboard can show progress %."""
+    import sql_console as s
+    d = _state["datasets"]
+    try:
+        if d["orders"]["total_source"] is None:
+            d["orders"]["total_source"] = s._table_row_count("postgres", "public.sales_orders")
+    except Exception:
+        pass
+    try:
+        if d["hr_events"]["total_source"] is None:
+            d["hr_events"]["total_source"] = s._table_row_count("mysql", "hr.employee_events")
+    except Exception:
+        pass
+    try:
+        if d["supply_events"]["total_source"] is None:
+            d["supply_events"]["total_source"] = s._table_row_count("mongodb", "supplychain.events")
+    except Exception:
+        pass
+
+
+def run_cycle() -> dict[str, Any]:
+    if _state["running_cycle"]:
+        return {"ok": True, "skipped": "cycle already running"}
+    _state["running_cycle"] = True
+    total = 0
+    cbytes = 0
+    order_rows = 0
+    revenue = 0.0
+    try:
+        _source_totals()
+        for d in DATASETS:
+            if not _state["enabled"]:
+                break
+            r = _offload_dataset(d["key"])
+            total += r["rows"]
+            cbytes += r["bytes"]
+            order_rows += r["order_rows"]
+            revenue += r["revenue"]
+        now = time.time()
+        _state["cycles"] += 1
+        _state["last_cycle_ts"] = now
+        _state["last_cycle_rows"] = total
+        _state["next_run_ts"] = now + float(_state["interval_s"])
+        if total:
+            minute = int(now // 60) * 60
+            if _state["minute"] and _state["minute"][-1][0] == minute:
+                m, r0, b0 = _state["minute"][-1]
+                _state["minute"][-1] = (minute, r0 + total, b0 + cbytes)
+            else:
+                _state["minute"].append((minute, total, cbytes))
+            point = {"t": _now().strftime("%H:%M:%S"), "rows": total, "bytes": cbytes,
+                     "orders": order_rows, "revenue": round(revenue, 2)}
+            _state["series"].append(point)
+            with _lock:
+                _business["ts"].append(point)
+            _feed(f"offloaded {total:,} rows → S3 Parquet · €{int(revenue):,} (cycle {_state['cycles']})")
+    finally:
+        _state["running_cycle"] = False
+    return {"ok": True, "rows": total, "bytes": cbytes, "revenue": round(revenue, 2)}
+
+
+def _loop() -> None:
+    _state["started_at"] = _now().isoformat()
+    time.sleep(12)  # let the API + sources settle
+    while True:
+        try:
+            if _state["enabled"]:
+                run_cycle()
+        except Exception:
+            pass
+        time.sleep(max(10.0, float(_state["interval_s"])))
+
+
+threading.Thread(target=_loop, daemon=True, name="etl-offload").start()
+
+
+# ── views ───────────────────────────────────────────────────────────────────────
+def _top(d: dict, n: int = 12, value: bool = False) -> list[dict]:
+    if value:
+        items = sorted(d.items(), key=lambda kv: -(kv[1].get("value", 0) if isinstance(kv[1], dict) else kv[1]))
+        return [{"key": k, "count": v.get("count", 0), "value": round(v.get("value", 0.0), 2)} for k, v in items[:n]]
+    items = sorted(d.items(), key=lambda kv: -kv[1])
+    return [{"key": k, "count": v} for k, v in items[:n]]
+
+
+def status_view() -> dict[str, Any]:
+    with _lock:
+        datasets = []
+        tot_parts = tot_rows = tot_bytes = 0
+        for d in DATASETS:
+            st = _state["datasets"][d["key"]]
+            total_src = st.get("total_source")
+            pct = None
+            if total_src and total_src > 0:
+                pct = min(100.0, round(100.0 * st["rows"] / total_src, 1))
+            tot_parts += st["parts"]
+            tot_rows += st["rows"]
+            tot_bytes += st["bytes"]
+            datasets.append({"key": d["key"], **{k: st[k] for k in (
+                "label", "engine", "color", "parts", "rows", "bytes", "backfilled",
+                "total_source", "last_ts", "last_rows", "last_key", "error")},
+                "progress_pct": pct})
+        now_min = int(time.time() // 60) * 60
+        rate = 0
+        for m, r, _b in _state["minute"]:
+            if m >= now_min - 60:
+                rate += r
+        return {
+            "ok": True,
+            "enabled": _state["enabled"],
+            "interval_s": _state["interval_s"],
+            "chunk": _state["chunk"],
+            "running_cycle": _state["running_cycle"],
+            "cycles": _state["cycles"],
+            "started_at": _state["started_at"],
+            "last_cycle_ts": _state["last_cycle_ts"],
+            "last_cycle_rows": _state["last_cycle_rows"],
+            "next_run_ts": _state["next_run_ts"],
+            "totals": {"parts": tot_parts, "rows": tot_rows, "bytes": tot_bytes},
+            "rate_rows_per_min": rate,
+            "datasets": datasets,
+            "series": list(_state["series"]),
+            "feed": list(_state["feed"])[:24],
+        }
+
+
+def business_view() -> dict[str, Any]:
+    with _lock:
+        o = _business["orders"]
+        hr = _business["hr"]
+        sup = _business["supply"]
+        tel = _business["telemetry"]
+        tel_metrics = sorted(
+            ({"key": k, "count": v["count"], "avg": round(v["sum"] / v["count"], 2) if v["count"] else 0}
+             for k, v in tel["by_metric"].items()), key=lambda x: -x["count"])[:12]
+        # combine into a region matrix across datasets
+        regions: dict[str, dict] = {}
+        for k, v in o["by_region"].items():
+            regions.setdefault(k, {})["orders"] = v.get("count", 0)
+            regions[k]["revenue"] = round(v.get("value", 0.0), 2)
+        for k, v in hr["by_region"].items():
+            regions.setdefault(k, {})["hr_events"] = v
+        for k, v in sup["by_region"].items():
+            regions.setdefault(k, {})["supply_events"] = v
+        region_matrix = sorted(
+            ({"region": k, "orders": v.get("orders", 0), "revenue": v.get("revenue", 0),
+              "hr_events": v.get("hr_events", 0), "supply_events": v.get("supply_events", 0)}
+             for k, v in regions.items() if k), key=lambda r: -(r["revenue"] or 0))
+        return {
+            "ok": True,
+            "generated_at": _now().isoformat(),
+            "kpis": {
+                "orders": o["count"], "revenue": round(o["revenue"], 2),
+                "avg_order": round(o["revenue"] / o["count"], 2) if o["count"] else 0,
+                "hr_events": hr["count"], "supply_events": sup["count"],
+                "supply_amount": round(sup["amount"], 2), "telemetry": tel["count"],
+                "rows_total": o["count"] + hr["count"] + sup["count"] + tel["count"],
+            },
+            "orders_by_region": _top(o["by_region"], 12, value=True),
+            "orders_by_status": _top(o["by_status"], 8),
+            "orders_by_channel": _top(o["by_channel"], 8, value=True),
+            "orders_by_currency": _top(o["by_currency"], 6),
+            "hr_by_department": _top(hr["by_department"], 10),
+            "hr_by_event": _top(hr["by_event"], 8),
+            "supply_by_type": _top(sup["by_type"], 10, value=True),
+            "telemetry_by_metric": tel_metrics,
+            "region_matrix": region_matrix,
+            "ts": list(_business["ts"]),
+        }
+
+
+@router.get("/status")
+async def get_status() -> JSONResponse:
+    return JSONResponse(status_view())
+
+
+@router.get("/business")
+async def get_business() -> JSONResponse:
+    return JSONResponse(business_view())
+
+
+@router.post("/run")
+async def post_run() -> JSONResponse:
+    from starlette.concurrency import run_in_threadpool
+    res = await run_in_threadpool(run_cycle)
+    return JSONResponse({**res, "status": status_view()})
+
+
+@router.post("/config")
+async def post_config(body: dict = Body(default={})) -> JSONResponse:
+    if "enabled" in body:
+        _state["enabled"] = bool(body["enabled"])
+    if "interval_s" in body:
+        try:
+            _state["interval_s"] = max(10.0, min(900.0, float(body["interval_s"])))
+        except Exception:
+            pass
+    if "chunk" in body:
+        try:
+            _state["chunk"] = max(200, min(50000, int(body["chunk"])))
+        except Exception:
+            pass
+    _feed(f"config updated · interval={_state['interval_s']}s · chunk={_state['chunk']} · enabled={_state['enabled']}")
+    return JSONResponse({"ok": True, "enabled": _state["enabled"],
+                         "interval_s": _state["interval_s"], "chunk": _state["chunk"]})

api/main.py

@@ -54,6 +54,7 @@ from streaming_ops import router as streaming_router
 from spark_workbench import router as spark_workbench_router
 from pii_catalog import router as pii_router
 from trino_federated import router as federated_router
+from etl_offload import router as etl_offload_router
 from ssh_terminal import ssh_session
 from node_registry import NODE_IDS, NODE_AGENT, NODE_REGISTRY, is_node_id
 from node_ops import build_node_detail, probe_node, run_node_probe_task
@@ -759,6 +760,7 @@ app.include_router(streaming_router)
 app.include_router(spark_workbench_router)
 app.include_router(pii_router)
 app.include_router(federated_router)
+app.include_router(etl_offload_router)
 app.add_middleware(
     CORSMiddleware,
     allow_origins=["*"],

api/requirements.txt

@@ -16,3 +16,4 @@ python-pptx==1.0.2
 boto3==1.35.99
 paramiko==3.5.0
 aiokafka==0.12.0
+pyarrow==18.1.0

api/storage_s3.py

@@ -2,7 +2,10 @@
 
 from __future__ import annotations
 
+import json as _json
 import os
+import random as _rnd
+import threading as _threading
 import time
 from collections import deque
 from datetime import datetime, timezone
@@ -34,6 +37,148 @@ _activity: deque[tuple[float, str]] = deque(maxlen=20000)
 _analytics_cache: dict[str, Any] = {"ts": 0.0, "data": None}
 _ANALYTICS_TTL = 45.0
 
+# ── Pipeline → S3 archiver ────────────────────────────────────────────────────
+# When data is generated, the streaming pipeline must actually land objects in
+# S3 so the dashboard reflects it (Last write / growth / activity). We mirror two
+# real stages straight into the object store:
+#   • Kafka → S3 CDC archive   -> cdc-archive/dt=YYYY-MM-DD/events-*.json (raw NDJSON)
+#   • Spark → S3 curated layer -> curated/sales_orders_masked/dt=…/part-*.json (PII masked)
+# Writes are buffered like a Kafka-Connect S3 sink (flush on size or interval) so
+# we don't create a flood of tiny objects, and force-flushed on a manual burst.
+S3_ARCHIVE_BUCKET = os.getenv("S3_ARCHIVE_BUCKET", "data")
+S3_FLUSH_INTERVAL = float(os.getenv("S3_ARCHIVE_FLUSH_S", "12"))
+S3_FLUSH_SIZE = int(os.getenv("S3_ARCHIVE_FLUSH_SIZE", "400"))
+
+_arch_lock = _threading.Lock()
+_arch_buf: dict[str, list] = {"cdc": [], "curated": []}
+_arch_since: dict[str, float] = {"t": 0.0}
+_last_write: dict[str, Any] = {}
+_arch_stats: dict[str, int] = {"objects": 0, "bytes": 0, "rows": 0}
+
+
+def _iso(v: Any) -> str:
+    return v.isoformat() if hasattr(v, "isoformat") else str(v)
+
+
+def _mask_cust(cid: Any) -> str:
+    h = abs(hash(("cust", cid))) % 0xFFFFFF
+    return f"cust_{h:06x}***"
+
+
+def _order_to_cdc(t: tuple, ts: str) -> dict[str, Any]:
+    cid, pid, region, channel, ots, amt, curr, status = t
+    return {"op": "c", "source": "postgres", "db": "sales", "table": "sales_orders", "ts": ts,
+            "after": {"customer_id": cid, "product_id": pid, "region": region, "sales_channel": channel,
+                      "amount": amt, "currency": curr, "order_status": status, "order_ts": _iso(ots)}}
+
+
+def _order_to_curated(t: tuple, ts: str) -> dict[str, Any]:
+    cid, pid, region, channel, ots, amt, curr, status = t
+    return {"customer_ref": _mask_cust(cid), "product_id": pid, "region": region, "sales_channel": channel,
+            "amount": amt, "currency": curr, "order_status": status, "order_ts": _iso(ots),
+            "ingested_ts": ts, "pii_masked": True}
+
+
+def _hr_to_cdc(t: tuple, ts: str) -> dict[str, Any]:
+    eid, dept, role, region, evt, sal, ets = t
+    return {"op": "c", "source": "mysql", "db": "hr", "table": "employee_events", "ts": ts,
+            "after": {"employee_id": eid, "department": dept, "role_name": role, "region": region,
+                      "event_type": evt, "salary_change": sal, "event_ts": _iso(ets)}}
+
+
+def _supply_to_cdc(d: dict, ts: str) -> dict[str, Any]:
+    return {"op": "c", "source": "mongodb", "db": "supplychain", "table": "events", "ts": ts, "after": dict(d)}
+
+
+def _tel_to_cdc(t: tuple, ts: str) -> dict[str, Any]:
+    dev, mts, mtype, mval, _payload = t
+    return {"op": "c", "source": "cassandra", "db": "telemetry", "table": "device_metrics", "ts": ts,
+            "after": {"device_id": dev, "metric_ts": _iso(mts), "metric_type": mtype, "metric_value": mval}}
+
+
+def _put(s3, bucket: str, key: str, body: bytes, content_type: str) -> None:
+    s3.put_object(Bucket=bucket, Key=key, Body=body, ContentType=content_type)
+    _track("write")
+    _last_write.update({"ts": datetime.now(timezone.utc).isoformat(), "mono": time.time(),
+                        "bucket": bucket, "key": key, "bytes": len(body)})
+    _arch_stats["objects"] += 1
+    _arch_stats["bytes"] += len(body)
+
+
+def _flush_locked() -> list[dict[str, Any]] | None:
+    cdc = _arch_buf["cdc"]
+    cur = _arch_buf["curated"]
+    if not cdc and not cur:
+        return None
+    s3 = _client()
+    now = datetime.now(timezone.utc)
+    day = now.strftime("%Y-%m-%d")
+    ms = int(now.timestamp() * 1000)
+    rid = _rnd.randint(1000, 9999)
+    written: list[dict[str, Any]] = []
+    if cdc:
+        body = ("\n".join(_json.dumps(e, default=str) for e in cdc) + "\n").encode()
+        key = f"cdc-archive/dt={day}/events-{ms}-{rid}.json"
+        _put(s3, S3_ARCHIVE_BUCKET, key, body, "application/x-ndjson")
+        written.append({"stage": "kafka→s3", "key": key, "rows": len(cdc), "bytes": len(body)})
+    if cur:
+        body = ("\n".join(_json.dumps(e, default=str) for e in cur) + "\n").encode()
+        key = f"curated/sales_orders_masked/dt={day}/part-{ms}-{rid}.json"
+        _put(s3, S3_ARCHIVE_BUCKET, key, body, "application/x-ndjson")
+        written.append({"stage": "spark→s3", "key": key, "rows": len(cur), "bytes": len(body)})
+    _arch_buf["cdc"] = []
+    _arch_buf["curated"] = []
+    _arch_since["t"] = time.time()
+    return written
+
+
+def archive_generated_batch(orders_rows=None, hr_rows=None, supply_docs=None, tel_rows=None,
+                            *, force: bool = False) -> list[dict[str, Any]] | dict[str, Any] | None:
+    """Stage a freshly generated batch into S3 (Kafka→S3 CDC archive + Spark→S3
+    curated masked). Buffered; flushes on size/interval or when force=True."""
+    try:
+        with _arch_lock:
+            ts = datetime.now(timezone.utc).isoformat()
+            for t in (orders_rows or []):
+                _arch_buf["cdc"].append(_order_to_cdc(t, ts))
+                _arch_buf["curated"].append(_order_to_curated(t, ts))
+                _arch_stats["rows"] += 1
+            for t in (hr_rows or []):
+                _arch_buf["cdc"].append(_hr_to_cdc(t, ts)); _arch_stats["rows"] += 1
+            for d in (supply_docs or []):
+                _arch_buf["cdc"].append(_supply_to_cdc(d, ts)); _arch_stats["rows"] += 1
+            for t in (tel_rows or []):
+                _arch_buf["cdc"].append(_tel_to_cdc(t, ts)); _arch_stats["rows"] += 1
+            if _arch_since["t"] == 0.0:
+                _arch_since["t"] = time.time()
+            buffered = len(_arch_buf["cdc"]) + len(_arch_buf["curated"])
+            age = time.time() - _arch_since["t"]
+            if force or buffered >= S3_FLUSH_SIZE or age >= S3_FLUSH_INTERVAL:
+                return _flush_locked()
+    except Exception as exc:  # never break the generator on an S3 hiccup
+        return {"error": str(exc)}
+    return None
+
+
+def put_object_bytes(key: str, body: bytes, content_type: str = "application/octet-stream",
+                     bucket: str | None = None) -> dict[str, Any]:
+    """Write raw bytes to S3 (used by the ETL offload agent for Parquet parts).
+    Tracks last-write + activity so the storage dashboard reflects it live."""
+    b = bucket or S3_ARCHIVE_BUCKET
+    s3 = _client()
+    _put(s3, b, key, body, content_type)
+    return {"ok": True, "bucket": b, "key": key, "bytes": len(body)}
+
+
+def archive_active(window_s: float = 25.0) -> bool:
+    """True if the pipeline wrote to S3 recently — drives the kafka→S3 edge pulse."""
+    return (time.time() - float(_last_write.get("mono") or 0.0)) < window_s
+
+
+def archive_info() -> dict[str, Any]:
+    return {"last_write": dict(_last_write) or None, "objects": _arch_stats["objects"],
+            "bytes": _arch_stats["bytes"], "rows": _arch_stats["rows"]}
+
 
 def _client():
     return boto3.client(
@@ -494,5 +639,18 @@ async def analytics(refresh: bool = Query(False)):
             _analytics_cache["ts"] = now
         except Exception as exc:
             return JSONResponse({"ok": False, "error": str(exc), "endpoint": S3_ENDPOINT}, status_code=502)
+
+    # Overlay the live last-write so the dashboard reflects pipeline writes
+    # immediately, without waiting for the (bounded, cached) full rescan.
+    summary = dict(data.get("summary") or {})
+    recent = list(data.get("recent") or [])
+    lw = dict(_last_write)
+    if lw.get("ts"):
+        if not summary.get("newest") or lw["ts"] > summary["newest"]:
+            summary["newest"] = lw["ts"]
+        recent = ([{"modified": lw["ts"], "bucket": lw.get("bucket"), "key": lw.get("key"),
+                    "bytes": lw.get("bytes", 0), "size_human": _human_size(lw.get("bytes", 0))}]
+                  + [r for r in recent if r.get("key") != lw.get("key")])[:15]
     return {"ok": True, "endpoint": S3_ENDPOINT, "generated_at": datetime.now(timezone.utc).isoformat(),
-            "activity": _activity_view(), **data}
+            "activity": _activity_view(), **data, "summary": summary, "recent": recent,
+            "archive": archive_info()}

api/trino_federated.py

@@ -453,25 +453,30 @@ _TEL_INSERT_TPL = ("INSERT INTO {ks}.device_metrics "
 def _gen_orders(n: int):
     rows, by_r, by_s, val = _order_rows(n)
     _gen_pg().cursor().executemany(_PG_INSERT, rows)
-    return by_r, by_s, val
+    return rows, by_r, by_s, val
 
 
 def _gen_hr(n: int):
-    _gen_mysql().cursor().executemany(_MYSQL_INSERT, _hr_rows(n))
+    rows = _hr_rows(n)
+    _gen_mysql().cursor().executemany(_MYSQL_INSERT, rows)
+    return rows
 
 
 def _gen_supply(n: int):
     docs = _supply_docs(n)
     if docs:
-        _gen_mongo()["events"].insert_many(docs)
+        _gen_mongo()["events"].insert_many([dict(d) for d in docs])
+    return docs
 
 
 def _gen_tel(n: int):
     import sql_console as s
     sess = _gen_cass()
     cql = _TEL_INSERT_TPL.format(ks=s.CASS_KS)
-    for row in _tel_rows(n):
+    rows = _tel_rows(n)
+    for row in rows:
         sess.execute(cql, row)
+    return rows
 
 
 def _generate_once(orders: int, hr: int, supply: int, tel: int) -> dict[str, Any]:
@@ -482,10 +487,15 @@ def _generate_once(orders: int, hr: int, supply: int, tel: int) -> dict[str, Any
     by_r: dict[str, int] = {}
     by_s: dict[str, int] = {}
     val = 0.0
+    order_built: list = []
+    hr_built: list = []
+    supply_built: list = []
+    tel_built: list = []
     if orders > 0:
         try:
             import psycopg2
             rows, by_r, by_s, val = _order_rows(orders)
+            order_built = rows
             c = psycopg2.connect(host=s.DB_HOST, port=s.PG_PORT, user=s.PG_USER, password=s.PG_PASS, dbname=s.PG_DB, connect_timeout=8)
             try:
                 c.autocommit = True
@@ -498,9 +508,10 @@ def _generate_once(orders: int, hr: int, supply: int, tel: int) -> dict[str, Any
     if hr > 0:
         try:
             import pymysql
+            hr_built = _hr_rows(hr)
             c = pymysql.connect(host=s.DB_HOST, port=s.MYSQL_PORT, user=s.MYSQL_USER, password=s.MYSQL_PASS, database=s.MYSQL_DB, connect_timeout=8, autocommit=True)
             try:
-                c.cursor().executemany(_MYSQL_INSERT, _hr_rows(hr))
+                c.cursor().executemany(_MYSQL_INSERT, hr_built)
                 out["hr_events"] = hr
             finally:
                 c.close()
@@ -508,9 +519,10 @@ def _generate_once(orders: int, hr: int, supply: int, tel: int) -> dict[str, Any
             pass
     if supply > 0:
         try:
+            supply_built = _supply_docs(supply)
             cli = s._mongo_client()
             try:
-                cli[s.MONGO_DB]["events"].insert_many(_supply_docs(supply))
+                cli[s.MONGO_DB]["events"].insert_many([dict(d) for d in supply_built])
                 out["supply_events"] = supply
             finally:
                 cli.close()
@@ -518,17 +530,28 @@ def _generate_once(orders: int, hr: int, supply: int, tel: int) -> dict[str, Any
             pass
     if tel > 0:
         try:
+            tel_built = _tel_rows(tel)
             cluster = s._cass_cluster()
             sess = cluster.connect()
             try:
                 cql = _TEL_INSERT_TPL.format(ks=s.CASS_KS)
-                for row in _tel_rows(tel):
+                for row in tel_built:
                     sess.execute(cql, row)
                 out["telemetry"] = tel
             finally:
                 cluster.shutdown()
         except Exception:
             pass
+    # Land the generated batch in S3 through the pipeline stages (Kafka→S3 CDC
+    # archive + Spark→S3 curated masked) so Object Storage reflects it at once.
+    try:
+        from storage_s3 import archive_generated_batch
+        archive_generated_batch(order_built if out["orders"] else [],
+                                hr_built if out["hr_events"] else [],
+                                supply_built if out["supply_events"] else [],
+                                tel_built if out["telemetry"] else [], force=True)
+    except Exception:
+        pass
     # fold into the live counters + feed so the dashboard reflects it instantly
     with _gen_lock:
         c = _GEN["counts"]
@@ -561,22 +584,32 @@ def _gen_tick():
     by_r: dict[str, int] = {}
     by_s: dict[str, int] = {}
     val = 0.0
+    order_built: list = []
+    hr_built: list = []
+    supply_built: list = []
+    tel_built: list = []
     try:
-        by_r, by_s, val = _gen_orders(no)
+        order_built, by_r, by_s, val = _gen_orders(no)
     except Exception:
         _gen_reset("pg"); no = 0
     try:
-        _gen_hr(nh)
+        hr_built = _gen_hr(nh)
     except Exception:
         _gen_reset("mysql"); nh = 0
     try:
-        _gen_supply(ns)
+        supply_built = _gen_supply(ns)
     except Exception:
         _gen_reset("mongo"); ns = 0
     try:
-        _gen_tel(nt)
+        tel_built = _gen_tel(nt)
     except Exception:
         _gen_reset("cass"); nt = 0
+    # stream the batch into S3 (buffered like a Kafka-Connect S3 sink)
+    try:
+        from storage_s3 import archive_generated_batch
+        archive_generated_batch(order_built, hr_built, supply_built, tel_built, force=False)
+    except Exception:
+        pass
     with _gen_lock:
         c = _GEN["counts"]
         c["orders"] += no