diff --git a/api/Dockerfile b/api/Dockerfile index cf89312..fba1167 100644 --- a/api/Dockerfile +++ b/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 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 diff --git a/api/agent_ops.py b/api/agent_ops.py new file mode 100644 index 0000000..7634b13 --- /dev/null +++ b/api/agent_ops.py @@ -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) diff --git a/api/main.py b/api/main.py index 301a54a..5bb2dec 100644 --- a/api/main.py +++ b/api/main.py @@ -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=["*"],