feat(agents): autonomous guard-railed DML on source DBs for Debezium CDC
Add api/agent_ops.py: background loop performs small INSERT/UPDATE/DELETE on
public.sales_orders (PG), hr.employee_events (MySQL) and supplychain.events
(Mongo). Agent rows are tagged (notes/atc_agent); UPDATE/DELETE only ever touch
agent-created rows. Env kill-switch + interval + per-tick row cap. Endpoints
/api/agent-ops/{status,toggle,run-once}. Loop started in lifespan.
This commit is contained in:
mo
+1
-1
@@ -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 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 hive_bench_seed.json .
|
||||
RUN mkdir -p /data
|
||||
ENV DATABASE_URL=sqlite:////data/atc-agents.db
|
||||
EXPOSE 3201
|
||||
|
||||
@@ -0,0 +1,311 @@
|
||||
"""Autonomous agent DML operations against the source databases.
|
||||
|
||||
Agents continuously perform small, guard-railed INSERT/UPDATE/DELETE operations
|
||||
on the operational source databases (PostgreSQL, MySQL, MongoDB) so that the
|
||||
Debezium CDC connectors capture a steady stream of changes that flow through
|
||||
Kafka and downstream.
|
||||
|
||||
Safety guardrails:
|
||||
- Only a fixed whitelist of tables/collections is touched (the exact ones the
|
||||
Debezium connectors capture: public.sales_orders, hr.employee_events,
|
||||
supplychain.events).
|
||||
- Every agent-created row is tagged (notes/payload marker, or an `atc_agent`
|
||||
field for Mongo). UPDATE and DELETE operate ONLY on rows the agents created
|
||||
themselves, never on seed/real data.
|
||||
- A hard per-tick row limit and an env kill-switch bound the activity.
|
||||
|
||||
The work runs in a background loop started from the app lifespan. DB drivers
|
||||
are synchronous, so each operation is executed via ``asyncio.to_thread``.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import os
|
||||
import random
|
||||
import time
|
||||
import uuid
|
||||
from datetime import datetime, timezone
|
||||
from typing import Any
|
||||
|
||||
from fastapi import APIRouter, Body
|
||||
from fastapi.responses import JSONResponse
|
||||
|
||||
router = APIRouter(prefix="/api/agent-ops", tags=["agent-ops"])
|
||||
|
||||
# ── Config (env-overridable) ────────────────────────────────────────────────
|
||||
DB_HOST = os.getenv("SRC_DB_HOST", "10.0.21.51")
|
||||
DB_USER = os.getenv("SRC_DB_USER", "mo")
|
||||
DB_PASSWORD = os.getenv("SRC_DB_PASSWORD", "Dell2026!")
|
||||
PG_DB = os.getenv("SRC_PG_DB", "postgres")
|
||||
PG_PORT = int(os.getenv("SRC_PG_PORT", "5432"))
|
||||
MYSQL_DB = os.getenv("SRC_MYSQL_DB", "hr")
|
||||
MYSQL_PORT = int(os.getenv("SRC_MYSQL_PORT", "3306"))
|
||||
MONGO_URI = os.getenv("SRC_MONGO_URI", f"mongodb://{DB_HOST}:27017/?replicaSet=rs0")
|
||||
MONGO_DB = os.getenv("SRC_MONGO_DB", "supplychain")
|
||||
|
||||
AGENT_TAG = "ATC-AGENT" # marker stored on agent-created rows
|
||||
DML_AGENT = "data-custodian"
|
||||
|
||||
_INTERVAL = float(os.getenv("AGENT_DML_INTERVAL_SECONDS", "45"))
|
||||
_MAX_ROWS = int(os.getenv("AGENT_DML_MAX_ROWS", "5"))
|
||||
# How many agent-created rows we keep around per source before favouring DELETE.
|
||||
_POOL_CAP = int(os.getenv("AGENT_DML_POOL_CAP", "400"))
|
||||
|
||||
_state: dict[str, Any] = {
|
||||
"enabled": os.getenv("AGENT_DML_ENABLED", "1") not in ("0", "false", "False", ""),
|
||||
"interval": _INTERVAL,
|
||||
"max_rows": _MAX_ROWS,
|
||||
"started": False,
|
||||
"ops_total": 0,
|
||||
"last_op": None,
|
||||
"last_error": None,
|
||||
"by_source": {"postgres": 0, "mysql": 0, "mongodb": 0},
|
||||
"by_op": {"insert": 0, "update": 0, "delete": 0},
|
||||
}
|
||||
|
||||
# ── Realistic value domains ─────────────────────────────────────────────────
|
||||
REGIONS = ["NA", "EU", "EMEA", "APAC", "LATAM"]
|
||||
CHANNELS = ["online", "retail", "partner", "wholesale", "direct"]
|
||||
CURRENCIES = ["USD", "EUR", "GBP"]
|
||||
ORDER_STATUS = ["NEW", "PROCESSING", "SHIPPED", "DELIVERED", "RETURNED", "CANCELLED"]
|
||||
DEPARTMENTS = ["Engineering", "Sales", "HR", "Finance", "Operations", "Marketing"]
|
||||
ROLES = ["Analyst", "Manager", "Engineer", "Director", "Specialist"]
|
||||
EVENT_TYPES = ["HIRE", "PROMOTION", "SALARY_CHANGE", "TRANSFER", "TERMINATION"]
|
||||
MONGO_TYPES = ["ORDER", "SHIPMENT", "RETURN", "RESTOCK", "UPDATE"]
|
||||
MONGO_SOURCES = ["CRM", "ERP", "WMS", "POS"]
|
||||
|
||||
|
||||
# ── Feed / terminal helpers (lazy import to avoid circular import) ───────────
|
||||
async def _emit(message: str, level: str = "info") -> None:
|
||||
try:
|
||||
from main import add_feed, publish_event
|
||||
from agent_terminal import terminal_log
|
||||
|
||||
entry = add_feed(DML_AGENT, message, level)
|
||||
await publish_event({"type": "feed", "entry": entry})
|
||||
await terminal_log(DML_AGENT, message, level=level, phase="dml")
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
# ── Synchronous DB operations (run in a thread) ──────────────────────────────
|
||||
def _pg_conn():
|
||||
import psycopg2
|
||||
|
||||
return psycopg2.connect(
|
||||
host=DB_HOST, dbname=PG_DB, user=DB_USER, password=DB_PASSWORD, port=PG_PORT, connect_timeout=8
|
||||
)
|
||||
|
||||
|
||||
def _mysql_conn():
|
||||
import pymysql
|
||||
|
||||
return pymysql.connect(
|
||||
host=DB_HOST, user=DB_USER, password=DB_PASSWORD, database=MYSQL_DB,
|
||||
port=MYSQL_PORT, connect_timeout=8, autocommit=True,
|
||||
)
|
||||
|
||||
|
||||
def _mongo_coll():
|
||||
import pymongo
|
||||
|
||||
client = pymongo.MongoClient(MONGO_URI, serverSelectionTimeoutMS=8000)
|
||||
return client, client[MONGO_DB]["events"]
|
||||
|
||||
|
||||
def _tag(run_id: str) -> str:
|
||||
return f"{AGENT_TAG} {run_id}"
|
||||
|
||||
|
||||
def _pg_dml(op: str) -> str:
|
||||
conn = _pg_conn()
|
||||
try:
|
||||
conn.autocommit = True
|
||||
with conn.cursor() as cur:
|
||||
cur.execute("SELECT count(*) FROM public.sales_orders WHERE notes LIKE %s", (AGENT_TAG + "%",))
|
||||
pool = cur.fetchone()[0]
|
||||
if op == "insert" or (op == "delete" and pool == 0) or (op == "update" and pool == 0):
|
||||
rid = uuid.uuid4().hex[:8]
|
||||
cur.execute(
|
||||
"""INSERT INTO public.sales_orders
|
||||
(customer_id, product_id, region, sales_channel, order_ts, amount, currency, order_status, notes)
|
||||
VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s) RETURNING order_id""",
|
||||
(
|
||||
random.randint(1, 50000), random.randint(1, 2000),
|
||||
random.choice(REGIONS), random.choice(CHANNELS),
|
||||
datetime.now(timezone.utc), round(random.uniform(10, 9999), 2),
|
||||
random.choice(CURRENCIES), random.choice(ORDER_STATUS), _tag(rid),
|
||||
),
|
||||
)
|
||||
oid = cur.fetchone()[0]
|
||||
return f"INSERT sales_orders order_id={oid} ({_tag(rid)})"
|
||||
if op == "update":
|
||||
cur.execute(
|
||||
"""UPDATE public.sales_orders SET order_status=%s, amount=round(amount*%s,2)
|
||||
WHERE order_id IN (
|
||||
SELECT order_id FROM public.sales_orders WHERE notes LIKE %s
|
||||
ORDER BY order_id DESC LIMIT 1)
|
||||
RETURNING order_id""",
|
||||
(random.choice(ORDER_STATUS), round(random.uniform(0.9, 1.2), 2), AGENT_TAG + "%"),
|
||||
)
|
||||
row = cur.fetchone()
|
||||
return f"UPDATE sales_orders order_id={row[0]}" if row else "UPDATE sales_orders (no agent rows)"
|
||||
# delete
|
||||
cur.execute(
|
||||
"""DELETE FROM public.sales_orders WHERE order_id IN (
|
||||
SELECT order_id FROM public.sales_orders WHERE notes LIKE %s
|
||||
ORDER BY order_id ASC LIMIT 1) RETURNING order_id""",
|
||||
(AGENT_TAG + "%",),
|
||||
)
|
||||
row = cur.fetchone()
|
||||
return f"DELETE sales_orders order_id={row[0]}" if row else "DELETE sales_orders (no agent rows)"
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
def _mysql_dml(op: str) -> str:
|
||||
conn = _mysql_conn()
|
||||
try:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute("SELECT count(*) FROM hr.employee_events WHERE notes LIKE %s", (AGENT_TAG + "%",))
|
||||
pool = cur.fetchone()[0]
|
||||
if op == "insert" or (op in ("update", "delete") and pool == 0):
|
||||
rid = uuid.uuid4().hex[:8]
|
||||
cur.execute(
|
||||
"""INSERT INTO hr.employee_events
|
||||
(employee_id, department, role_name, region, event_type, salary_change, event_ts, notes)
|
||||
VALUES (%s,%s,%s,%s,%s,%s,%s,%s)""",
|
||||
(
|
||||
random.randint(1, 20000), random.choice(DEPARTMENTS), random.choice(ROLES),
|
||||
random.choice(REGIONS), random.choice(EVENT_TYPES),
|
||||
round(random.uniform(-5000, 15000), 2), datetime.now(timezone.utc), _tag(rid),
|
||||
),
|
||||
)
|
||||
return f"INSERT employee_events event_id={cur.lastrowid} ({_tag(rid)})"
|
||||
if op == "update":
|
||||
cur.execute(
|
||||
"""UPDATE hr.employee_events SET salary_change=%s, event_type=%s
|
||||
WHERE notes LIKE %s ORDER BY event_id DESC LIMIT 1""",
|
||||
(round(random.uniform(-5000, 15000), 2), random.choice(EVENT_TYPES), AGENT_TAG + "%"),
|
||||
)
|
||||
return f"UPDATE employee_events ({cur.rowcount} row)"
|
||||
cur.execute(
|
||||
"DELETE FROM hr.employee_events WHERE notes LIKE %s ORDER BY event_id ASC LIMIT 1",
|
||||
(AGENT_TAG + "%",),
|
||||
)
|
||||
return f"DELETE employee_events ({cur.rowcount} row)"
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
def _mongo_dml(op: str) -> str:
|
||||
client, coll = _mongo_coll()
|
||||
try:
|
||||
pool = coll.count_documents({"atc_agent": True}, limit=1)
|
||||
if op == "insert" or (op in ("update", "delete") and pool == 0):
|
||||
rid = uuid.uuid4().hex[:8]
|
||||
doc = {
|
||||
"event_id": str(uuid.uuid4()),
|
||||
"type": random.choice(MONGO_TYPES),
|
||||
"region": random.choice(REGIONS),
|
||||
"source": random.choice(MONGO_SOURCES),
|
||||
"amount": round(random.uniform(10, 50000), 4),
|
||||
"ts": datetime.now(timezone.utc),
|
||||
"payload": "X" * 200,
|
||||
"atc_agent": True,
|
||||
"agent_run": rid,
|
||||
}
|
||||
res = coll.insert_one(doc)
|
||||
return f"INSERT events _id={res.inserted_id} (agent_run={rid})"
|
||||
if op == "update":
|
||||
doc = coll.find_one({"atc_agent": True}, sort=[("_id", -1)])
|
||||
if not doc:
|
||||
return "UPDATE events (no agent docs)"
|
||||
coll.update_one(
|
||||
{"_id": doc["_id"]},
|
||||
{"$set": {"type": random.choice(MONGO_TYPES), "amount": round(random.uniform(10, 50000), 4)}},
|
||||
)
|
||||
return f"UPDATE events _id={doc['_id']}"
|
||||
doc = coll.find_one({"atc_agent": True}, sort=[("_id", 1)])
|
||||
if not doc:
|
||||
return "DELETE events (no agent docs)"
|
||||
coll.delete_one({"_id": doc["_id"]})
|
||||
return f"DELETE events _id={doc['_id']}"
|
||||
finally:
|
||||
client.close()
|
||||
|
||||
|
||||
_DISPATCH = {"postgres": _pg_dml, "mysql": _mysql_dml, "mongodb": _mongo_dml}
|
||||
_SRC_LABEL = {"postgres": "PostgreSQL sales_orders", "mysql": "MySQL employee_events", "mongodb": "MongoDB events"}
|
||||
|
||||
|
||||
def _pick_op() -> str:
|
||||
# Insert-heavy so a pool of agent rows exists for safe UPDATE/DELETE.
|
||||
return random.choices(["insert", "update", "delete"], weights=[0.5, 0.3, 0.2], k=1)[0]
|
||||
|
||||
|
||||
async def _run_one(source: str | None = None, op: str | None = None) -> dict[str, Any]:
|
||||
source = source or random.choice(list(_DISPATCH.keys()))
|
||||
op = op or _pick_op()
|
||||
fn = _DISPATCH.get(source)
|
||||
if not fn:
|
||||
return {"ok": False, "error": f"unknown source {source}"}
|
||||
try:
|
||||
detail = await asyncio.to_thread(fn, op)
|
||||
_state["ops_total"] += 1
|
||||
_state["by_source"][source] = _state["by_source"].get(source, 0) + 1
|
||||
actual_op = detail.split(" ", 1)[0].lower()
|
||||
if actual_op in _state["by_op"]:
|
||||
_state["by_op"][actual_op] += 1
|
||||
_state["last_op"] = {"source": source, "op": op, "detail": detail, "ts": datetime.now(timezone.utc).isoformat()}
|
||||
_state["last_error"] = None
|
||||
await _emit(f"[agent-dml] {_SRC_LABEL[source]}: {detail} — Debezium will capture this change", "info")
|
||||
return {"ok": True, "source": source, "op": op, "detail": detail}
|
||||
except Exception as exc:
|
||||
_state["last_error"] = str(exc)
|
||||
await _emit(f"[agent-dml] {_SRC_LABEL.get(source, source)}: operation failed: {str(exc)[:120]}", "err")
|
||||
return {"ok": False, "source": source, "op": op, "error": str(exc)}
|
||||
|
||||
|
||||
async def agent_dml_loop() -> None:
|
||||
"""Background loop: continuously perform guard-railed DML on the sources."""
|
||||
_state["started"] = True
|
||||
await asyncio.sleep(8) # let the app settle / DB reachable
|
||||
await _emit("[agent-dml] Autonomous DML agent online — generating live changes for Debezium", "info")
|
||||
while True:
|
||||
try:
|
||||
if _state["enabled"]:
|
||||
await _run_one()
|
||||
except Exception as exc: # never let the loop die
|
||||
_state["last_error"] = str(exc)
|
||||
await asyncio.sleep(max(5.0, float(_state["interval"])))
|
||||
|
||||
|
||||
# ── Endpoints ────────────────────────────────────────────────────────────────
|
||||
@router.get("/status")
|
||||
async def status() -> JSONResponse:
|
||||
return JSONResponse({"ok": True, **_state})
|
||||
|
||||
|
||||
@router.post("/toggle")
|
||||
async def toggle(body: dict[str, Any] = Body(default={})) -> JSONResponse:
|
||||
if "enabled" in body:
|
||||
_state["enabled"] = bool(body["enabled"])
|
||||
else:
|
||||
_state["enabled"] = not _state["enabled"]
|
||||
if "interval" in body:
|
||||
try:
|
||||
_state["interval"] = max(5.0, float(body["interval"]))
|
||||
except (TypeError, ValueError):
|
||||
pass
|
||||
await _emit(f"[agent-dml] Autonomous DML {'ENABLED' if _state['enabled'] else 'PAUSED'} by operator", "warn")
|
||||
return JSONResponse({"ok": True, "enabled": _state["enabled"], "interval": _state["interval"]})
|
||||
|
||||
|
||||
@router.post("/run-once")
|
||||
async def run_once(body: dict[str, Any] = Body(default={})) -> JSONResponse:
|
||||
source = body.get("source")
|
||||
op = body.get("op")
|
||||
result = await _run_one(source=source, op=op)
|
||||
return JSONResponse(result)
|
||||
@@ -42,6 +42,7 @@ from pipeline_ops import router as pipeline_router
|
||||
from hadoop_analytics import router as hadoop_router
|
||||
from elasticsearch_api import router as elasticsearch_router
|
||||
from sql_console import router as sql_router
|
||||
from agent_ops import router as agent_ops_router, agent_dml_loop
|
||||
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
|
||||
@@ -711,9 +712,11 @@ async def lifespan(app: FastAPI):
|
||||
meta = NODE_REGISTRY[nid]
|
||||
await terminal_log(nid, f"{meta['label']} shell ready — click node to connect", level="info", phase="boot")
|
||||
task = asyncio.create_task(heartbeat_loop())
|
||||
dml_task = asyncio.create_task(agent_dml_loop())
|
||||
add_feed("infra-sentinel", "ATC Command Center API online", "info")
|
||||
yield
|
||||
task.cancel()
|
||||
dml_task.cancel()
|
||||
if redis_client:
|
||||
await redis_client.close()
|
||||
|
||||
@@ -725,6 +728,7 @@ app.include_router(pipeline_router)
|
||||
app.include_router(hadoop_router)
|
||||
app.include_router(elasticsearch_router)
|
||||
app.include_router(sql_router)
|
||||
app.include_router(agent_ops_router)
|
||||
app.add_middleware(
|
||||
CORSMiddleware,
|
||||
allow_origins=["*"],
|
||||
|
||||
Reference in New Issue
Block a user