ATC/atc-agents
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add Command Center v2: DQ/RAG integration, S3 browser, Jupyter, GPU matrix.

Browse Source 
Mirror mo/atc-GPU layout with config/, docs/, scripts/ for Gitea deploy.
This commit is contained in:
mo
2026-06-25 00:28:23 +00:00
parent fb9cc21c9a
commit a11621b21f
110 changed files with 14622 additions and 529 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/main.py
+705 -69
View File
@@ -5,6 +5,7 @@ from __future__ import annotations
import asyncio
import json
import os
import time
import uuid
from contextlib import asynccontextmanager
from datetime import datetime, timezone
@@ -12,15 +13,54 @@ from typing import Any
import httpx
import redis.asyncio as aioredis
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi import FastAPI, File, UploadFile, WebSocket, WebSocketDisconnect
from fastapi.middleware.cors import CORSMiddleware
from agent_terminal import (
get_all_terminals,
get_terminal_lines,
init_terminals,
make_logger,
set_terminal_publisher,
terminal_log,
)
from lab_context import collect_full_lab_context, format_context_for_agent
from presentation import build_presentation_payload, render_presentation_html
from presentation_upload import get_deck, list_decks, save_upload
from presentation_static import get_static_deck, list_static_decks
from storage_s3 import router as storage_s3_router
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
from approval_service import (
APPROVAL_ACTION_TYPES,
approval_stats,
create_approval_request,
decide_approval_request,
detect_agent_proposed_action,
detect_approval_intent,
list_approvals,
)
from db import SessionLocal, db_health, init_database
from supervisor import mirror_terminal_line, mirror_to_supervisors
from workload import build_workload_payload
_workload_cache: dict[str, Any] = {"ts": 0.0, "data": None}
_presentation_cache: dict[str, Any] = {"ts": 0.0, "data": None}
WORKLOAD_CACHE_TTL = 30.0
PRESENTATION_CACHE_TTL = 45.0
from pydantic import BaseModel, Field
from sqlalchemy import Column, DateTime, String, Text, create_engine, select
from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker
from sqlalchemy import Column, DateTime, String, Text, select
from sqlalchemy.orm import DeclarativeBase
REDIS_URL = os.getenv("REDIS_URL", "redis://redis:6379/0")
DOCKHAND_URL = os.getenv("DOCKHAND_URL", "http://10.0.21.45:8082")
DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:////data/atc-agents.db")
GPU_URL = os.getenv("GPU_URL", "http://10.0.20.106:9000")
GPU_UI_URL = os.getenv("GPU_UI_URL", GPU_URL)
LLM_URL = os.getenv("LLM_URL", "http://10.0.20.106:8001/v1")
LLM_MODEL = os.getenv("LLM_MODEL", "gpt-4o")
LLM_API_KEY = os.getenv("LLM_API_KEY", "sk-local")
LLM_TIMEOUT = float(os.getenv("LLM_TIMEOUT", "120"))
AGENTS = [
{
@@ -29,6 +69,14 @@ AGENTS = [
"color": "#00f0ff",
"zone": "etl",
"role": "Airflow, Kafka, Debezium, S3 pipeline",
"icon": "⚡",
"motto": "Pipelines never sleep",
"capabilities": ["Airflow", "Kafka", "Debezium", "S3", "Connectors"],
"suggested_prompts": [
"How is Debezium doing?",
"Are all Airflow DAGs healthy?",
"Kafka connector status?",
],
},
{
"id": "lakehouse-ops",
@@ -36,6 +84,14 @@ AGENTS = [
"color": "#ff00aa",
"zone": "lakehouse",
"role": "Spark, Trino, Iceberg",
"icon": "🏔️",
"motto": "Query the lake, trust the table",
"capabilities": ["Spark", "Trino", "Iceberg", "Delta", "SQL"],
"suggested_prompts": [
"Lakehouse stack status?",
"Is Trino reachable?",
"How many lakehouse containers are running?",
],
},
{
"id": "data-custodian",
@@ -43,6 +99,14 @@ AGENTS = [
"color": "#ffaa00",
"zone": "db",
"role": "PostgreSQL, MySQL, Mongo, Cassandra, Neo4j",
"icon": "🛡️",
"motto": "Guardian of every row",
"capabilities": ["PostgreSQL", "MySQL", "MongoDB", "Cassandra", "Neo4j"],
"suggested_prompts": [
"Hoeveel data zit er in de databases?",
"Wat staat er in PostgreSQL sales_orders?",
"MongoDB supplychain overzicht",
],
},
{
"id": "hadoop-ranger",
@@ -50,13 +114,93 @@ AGENTS = [
"color": "#39ff14",
"zone": "hadoop",
"role": "HDFS, YARN cluster",
"icon": "🌲",
"motto": "Patrol the data forest",
"capabilities": ["HDFS", "YARN", "NameNode", "DataNodes"],
"suggested_prompts": [
"Is HDFS NameNode up?",
"Hadoop cluster status?",
"YARN nodes healthy?",
],
},
{
"id": "infra-sentinel",
"name": "Infra Sentinel",
"color": "#b366ff",
"color": "#9b72cf",
"zone": "docker",
"role": "Docker, Proxmox, monitoring",
"role": "Docker, Proxmox, GPU, monitoring",
"icon": "👁️",
"motto": "See everything, miss nothing",
"capabilities": ["Docker", "Proxmox", "GPU", "vLLM", "Monitoring"],
"suggested_prompts": [
"GPU status?",
"Which LLM model is running?",
"Docker container overview",
],
},
{
"id": "mo-commander",
"name": "Mo · Command",
"color": "#4c9aed",
"zone": "command",
"role": "Supervisor — full event intel, ingress, approvals",
"icon": "🎯",
"motto": "Nothing happens without Mo knowing",
"supervisor": True,
"person": "mo",
"capabilities": ["Events", "Ingress", "Approvals", "Agent dispatch", "Network IN"],
"suggested_prompts": [
"What happened today?",
"What events came in?",
"Pipeline status overview",
],
},
{
"id": "bart-commander",
"name": "Bart · Ops",
"color": "#3fb950",
"zone": "command",
"role": "Supervisor — egress, MCP comms, network OUT",
"icon": "📡",
"motto": "All traffic flows through Bart",
"supervisor": True,
"person": "bart",
"capabilities": ["Egress", "MCP routing", "Network OUT", "GPU inference", "S3 writes"],
"suggested_prompts": [
"What is leaving the cluster?",
"MCP agent communication status?",
"Network egress overview",
],
},
{
"id": "network-watcher",
"name": "Network Watcher",
"color": "#58a6ff",
"zone": "network",
"role": "VLAN 20/21 traffic, data in & out paths",
"icon": "🌐",
"motto": "Every packet tells a story",
"capabilities": ["VLAN 20", "VLAN 21", "Ingress", "Egress", "Firewall paths"],
"suggested_prompts": [
"Data ingress status?",
"What leaves the cluster?",
"Network path to S3?",
],
},
{
"id": "mcp-coordinator",
"name": "MCP Coordinator",
"color": "#f778ba",
"zone": "mcp",
"role": "MCP hub — routes all agent tool calls & comms",
"icon": "🔀",
"motto": "Route once, deliver everywhere",
"capabilities": ["MCP servers", "Tool routing", "Agent relay", "WebSocket bus"],
"suggested_prompts": [
"Which MCP agents are active?",
"MCP hub route status?",
"Agent communication overview",
],
},
]
@@ -69,11 +213,18 @@ ZONES = [
]
INTENT_KEYWORDS: dict[str, list[str]] = {
"data-custodian": ["database", "db", "postgres", "mysql", "mongo", "cassandra", "neo4j", "sql"],
"lakehouse-ops": ["trino", "spark", "lakehouse", "iceberg", "query"],
"hadoop-ranger": ["hadoop", "hdfs", "yarn", "datanode"],
"infra-sentinel": ["docker", "container", "vm", "proxmox", "infra", "grafana"],
"data-custodian": ["database", "postgres", "postgresql", "mysql", "mongo", "mongodb", "cassandra", "neo4j", "sql", "db "],
"lakehouse-ops": ["trino", "spark", "lakehouse", "iceberg", "query", "table"],
"hadoop-ranger": [
"hadoop", "hdfs", "yarn", "datanode", "namenode", "replicatie", "replication",
"rf factor", "opslag", "bestanden", "blocks", "cluster opslag", "data op",
],
"infra-sentinel": ["docker", "container", "vm", "proxmox", "infra", "grafana", "gpu", "vllm", "llm", "nvidia", "inference", "model"],
"etl-guardian": ["airflow", "dag", "debezium", "kafka", "connector", "etl", "pipeline", "s3"],
"network-watcher": ["network", "vlan", "ingress", "egress", "traffic", "packet", "firewall", "route"],
"mcp-coordinator": ["mcp", "tool", "router", "relay", "websocket", "hub"],
"mo-commander": ["mo", "supervisor", "events", "overzicht", "alles", "gebeurd"],
"bart-commander": ["bart", "egress", "uitgaand", "communicatie", "mcp comm"],
}
@@ -100,11 +251,16 @@ class Approval(Base):
action = Column(Text)
reason = Column(Text)
status = Column(String, default="pending")
action_type = Column(String, default="generic.mutate")
target = Column(Text, default="")
payload = Column(Text, default="{}")
decided_by = Column(String, nullable=True)
decide_note = Column(Text, nullable=True)
decided_at = Column(DateTime, nullable=True)
priority = Column(String, default="normal")
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(bind=engine)
Base.metadata.create_all(engine)
_db_info = init_database(Base)
redis_client: aioredis.Redis | None = None
ws_clients: set[WebSocket] = set()
@@ -112,14 +268,35 @@ ws_clients: set[WebSocket] = set()
class PromptRequest(BaseModel):
message: str = Field(min_length=1, max_length=2000)
agent_id: str | None = None
class NodeAskRequest(BaseModel):
message: str = Field(min_length=1, max_length=2000)
class ApprovalCreateRequest(BaseModel):
agent_id: str = Field(min_length=1, max_length=64)
action: str = Field(min_length=1, max_length=2000)
reason: str = Field(min_length=1, max_length=2000)
action_type: str = "generic.mutate"
target: str = ""
payload: dict[str, Any] | None = None
priority: str = "normal"
class ApprovalDecision(BaseModel):
approved: bool
decided_by: str = "mo-commander"
note: str = ""
def route_agent(message: str) -> str:
lower = message.lower()
# Storage/data questions default to Hadoop unless clearly about databases
if any(w in lower for w in ("data", "opslag", "gb", "replicatie", "replication", "hdfs", "hadoop")):
if not any(w in lower for w in ("postgres", "mysql", "mongo", "database", "sql", "neo4j", "cassandra")):
return "hadoop-ranger"
scores = {aid: sum(1 for kw in kws if kw in lower) for aid, kws in INTENT_KEYWORDS.items()}
best = max(scores, key=scores.get)
if scores[best] == 0:
@@ -127,6 +304,94 @@ def route_agent(message: str) -> str:
return best
async def gather_agent_context(
agent_id: str,
status: dict[str, Any],
log: Any | None = None,
) -> str:
"""Full lab snapshot for vLLM — all domains, agent's primary domain highlighted."""
snapshot = await collect_full_lab_context(gpu_data=status.get("gpu"), log=log)
snapshot["domains_summary"] = status.get("domains", {})
ctx = format_context_for_agent(agent_id, snapshot)
agent_lines = ["", "=== AGENTS & SUPERVISORS ==="]
for a in AGENTS:
sup = " [supervisor]" if a.get("supervisor") else ""
agent_lines.append(f" - {a['name']} ({a['id']}){sup}: {a['role']}")
ctx = ctx + "\n".join(agent_lines)
if log:
await log("info", "fetch", f"▸ Context assembled: {len(ctx)} chars for LLM")
return ctx
async def ask_llm(
agent_id: str,
message: str,
context: str,
log: Any | None = None,
) -> str | None:
agent = next(a for a in AGENTS if a["id"] == agent_id)
system = f"""Je bent {agent['name']}, een autonomous ops agent in het Dell ATC data lab.
Specialisatie: {agent['role']}.
Motto: {agent.get('motto', '')}
Je antwoordt namens je domein maar hebt zicht op de HELE lab stack: Docker, databases, lakehouse (Trino/Spark/Kafka Connect), ETL (Airflow/Kafka), Hadoop HDFS, en GPU/vLLM.
Regels:
- Antwoord in dezelfde taal als de gebruiker (Nederlands of Engels).
- Je hebt volledige zicht op de HELE cluster: alle VMs, zones, connectors, GPU, Hadoop, ObjectScale en Command Center.
- Gebruik ALLEEN de live data hieronder — verzin geen hosts, poorten, cijfers of connector namen.
- Gebruik exact de container/connector namen uit de data (bijv. mysql-hr-connector, niet "Debezium").
- Als iets DOWN of 0 GB is, zeg dat eerlijk.
- Kort en behulpzaam (max ~10 zinnen); bullet lists mogen als het overzicht helpt.
--- LIVE LAB DATA (primary domain eerst, daarna volledige stack) ---
{context}
"""
if log:
await log("info", "llm", f"▸ Querying vLLM model={LLM_MODEL}")
await log("cmd", "llm", f"$ POST {LLM_URL.rstrip('/')}/chat/completions")
await log("info", "llm", f" user: {message[:160]}{'…' if len(message) > 160 else ''}")
try:
async with httpx.AsyncClient(timeout=LLM_TIMEOUT) as client:
t0 = time.monotonic()
r = await client.post(
f"{LLM_URL.rstrip('/')}/chat/completions",
headers={
"Authorization": f"Bearer {LLM_API_KEY}",
"Content-Type": "application/json",
},
json={
"model": LLM_MODEL,
"messages": [
{"role": "system", "content": system},
{"role": "user", "content": message},
],
"max_tokens": 800,
"temperature": 0.25,
},
)
r.raise_for_status()
content = r.json()["choices"][0]["message"]["content"].strip()
ms = int((time.monotonic() - t0) * 1000)
if content and content.strip("!"):
if log:
await log("ok", "llm", f"← vLLM response {len(content)} chars ({ms}ms)")
preview = content.replace("\n", " ")[:180]
await log("info", "llm", f" » {preview}{'…' if len(content) > 180 else ''}")
return content
if log:
await log("warn", "llm", f"← Empty or invalid LLM output ({ms}ms)")
except Exception as exc:
if log:
await log("err", "llm", f"✗ vLLM error: {exc}")
return None
def fallback_answer(agent_id: str, context: str) -> str:
agent_name = next(a["name"] for a in AGENTS if a["id"] == agent_id)
return f"**{agent_name}** (offline LLM — ruwe data):\n\n{context}"
async def publish_event(event: dict[str, Any]) -> None:
payload = json.dumps(event, default=str)
if redis_client:
@@ -140,6 +405,21 @@ async def publish_event(event: dict[str, Any]) -> None:
for ws in dead:
ws_clients.discard(ws)
et = event.get("type")
if et == "feed":
entry = event.get("entry") or {}
await mirror_to_supervisors(
entry.get("agent_id", "?"),
entry.get("message", ""),
level=entry.get("level", "info"),
)
elif et == "terminal":
await mirror_terminal_line(event.get("line") or {})
elif et in ("agent_dispatch", "agent_fetch", "agent_return"):
aid = event.get("agent_id", "?")
zone = event.get("zone", "")
await mirror_to_supervisors(aid, f"{et} → zone {zone}", level="info", phase="dispatch")
def add_feed(agent_id: str, message: str, level: str = "info") -> dict:
entry_id = str(uuid.uuid4())[:8]
@@ -175,6 +455,67 @@ async def probe_url(url: str) -> bool:
return False
async def collect_gpu() -> dict[str, Any]:
host = GPU_URL.replace("http://", "").replace("https://", "").split("/")[0]
base = {"ok": False, "host": host, "ui_url": GPU_UI_URL}
try:
async with httpx.AsyncClient(timeout=6.0) as client:
metrics_r, model_r, integration_r = await asyncio.gather(
client.get(f"{GPU_URL}/api/gpu/metrics"),
client.get(f"{GPU_URL}/api/active-model"),
client.get(f"{GPU_URL}/api/integration"),
return_exceptions=True,
)
gpus: list[dict[str, Any]] = []
if isinstance(metrics_r, httpx.Response) and metrics_r.status_code == 200:
current = metrics_r.json().get("current", {})
gpus = [
{
"index": g["index"],
"name": g["name"],
"util_gpu": g.get("util_gpu", 0),
"memory_used_mib": g.get("memory_used_mib", 0),
"memory_total_mib": g.get("memory_total_mib", 0),
"temperature_c": g.get("temperature_c", 0),
"power_w": g.get("power_w", 0),
}
for g in current.get("gpus", [])
]
active_model = None
inference_active = False
vllm_url = None
if isinstance(model_r, httpx.Response) and model_r.status_code == 200:
model_data = model_r.json()
active_model = model_data.get("name")
inference_active = bool(model_data.get("inference_active"))
vllm_url = model_data.get("base_url")
if isinstance(integration_r, httpx.Response) and integration_r.status_code == 200:
integ = integration_r.json()
if not active_model:
active_model = integ.get("active_name")
if not inference_active:
inference_active = bool(integ.get("inference_active"))
if not vllm_url:
vllm_url = integ.get("recommended_base_url")
return {
**base,
"ok": len(gpus) > 0 or inference_active,
"inference_active": inference_active,
"active_model": active_model,
"vllm_url": vllm_url,
"gpu_count": len(gpus),
"gpus": gpus,
}
except Exception as exc:
return {**base, "error": str(exc)}
async def collect_status() -> dict[str, Any]:
db_containers = await dockhand_env_containers(5)
db_running = sum(1 for c in db_containers if c.get("state") == "running")
@@ -201,6 +542,10 @@ async def collect_status() -> dict[str, Any]:
return "warn"
return "down"
gpu = await collect_gpu()
gpu_level = "ok" if gpu.get("ok") and gpu.get("inference_active") else ("warn" if gpu.get("ok") else "down")
gpu_label = gpu.get("active_model") or (f"{gpu.get('gpu_count', 0)} GPUs" if gpu.get("ok") else "offline")
return {
"ts": datetime.now(timezone.utc).isoformat(),
"domains": {
@@ -209,47 +554,112 @@ async def collect_status() -> dict[str, Any]:
"lakehouse": {"level": level(lake_running, lake_total), "label": f"{lake_running}/{lake_total} up", "running": lake_running, "total": lake_total},
"hadoop": {"level": "ok" if hdfs_ok else "warn", "label": "NN up" if hdfs_ok else "NN check"},
"etl": {"level": "ok" if kafka_ok and airflow_ok else "warn", "label": "Kafka+Airflow"},
"gpu": {"level": gpu_level, "label": gpu_label},
},
"gpu": gpu,
"kafka_ok": kafka_ok,
"airflow_ok": airflow_ok,
"hdfs_ok": hdfs_ok,
}
async def _run_agent_task_safe(agent_id: str, message: str, prompt_id: str) -> None:
try:
await run_agent_task(agent_id, message, prompt_id)
except Exception as exc:
agent_name = next((a["name"] for a in AGENTS if a["id"] == agent_id), agent_id)
err = f"Sorry — {agent_name} could not complete your request: {exc}"
await terminal_log(agent_id, f"[{prompt_id}] ✗ Error: {exc}", level="err", phase="error", prompt_id=prompt_id)
feed = add_feed(agent_id, f"{agent_name} failed: {str(exc)[:80]}", "err")
await publish_event({"type": "feed", "entry": feed})
await publish_event({
"type": "prompt_result",
"prompt_id": prompt_id,
"agent_id": agent_id,
"answer": err,
})
async def run_agent_task(agent_id: str, message: str, prompt_id: str) -> str:
zone = next(a["zone"] for a in AGENTS if a["id"] == agent_id)
agent_name = next(a["name"] for a in AGENTS if a["id"] == agent_id)
log = make_logger(agent_id, prompt_id)
approval_created = False
intent = detect_approval_intent(message)
if intent:
with SessionLocal() as db:
await create_approval_request(
db=db,
ApprovalModel=Approval,
agent_id=agent_id,
action=intent["action"],
reason=intent["reason"],
action_type=intent["action_type"],
terminal_log=terminal_log,
mirror_supervisors=mirror_to_supervisors,
publish=publish_event,
add_feed=add_feed,
)
approval_created = True
await terminal_log(
agent_id,
f"[{prompt_id}] Mutating request detected — approval queued for Mo & Bart",
level="warn",
phase="approval",
prompt_id=prompt_id,
)
await terminal_log(
agent_id,
f"[{prompt_id}] ▶ Mission accepted: {message}",
level="info",
phase="dispatch",
prompt_id=prompt_id,
)
await publish_event({"type": "agent_dispatch", "agent_id": agent_id, "zone": zone, "prompt_id": prompt_id})
await asyncio.sleep(0.8)
await asyncio.sleep(0.4)
await terminal_log(agent_id, f"[{prompt_id}] Walking to zone: {zone}", level="info", phase="dispatch", prompt_id=prompt_id)
await publish_event({"type": "agent_fetch", "agent_id": agent_id, "zone": zone, "prompt_id": prompt_id})
await log("info", "fetch", f"[{prompt_id}] Collecting live lab metrics…")
status = await collect_status()
answer_parts = [f"**{next(a['name'] for a in AGENTS if a['id'] == agent_id)}** reporting:"]
context = await gather_agent_context(agent_id, status, log=log)
if agent_id == "data-custodian":
db = status["domains"]["databases"]
containers = await dockhand_env_containers(5)
names = ", ".join(f"{c['name']}:{c.get('state','?')}" for c in containers[:8])
answer_parts.append(f"Databases {db['label']}. Containers: {names or 'unreachable'}.")
elif agent_id == "etl-guardian":
answer_parts.append(
f"Kafka UI: {'OK' if status['kafka_ok'] else 'DOWN'}. "
f"Airflow: {'OK' if status['airflow_ok'] else 'DOWN'}. "
f"Lakehouse {status['domains']['lakehouse']['label']}."
)
elif agent_id == "lakehouse-ops":
answer_parts.append(f"Lakehouse stack {status['domains']['lakehouse']['label']}. Trino at 10.0.21.50:8089.")
elif agent_id == "hadoop-ranger":
answer_parts.append(f"HDFS NameNode: {'reachable' if status['hdfs_ok'] else 'unreachable'} on 10.0.21.61:9870.")
else:
answer_parts.append(
f"Docker {status['domains']['docker']['label']}. "
f"Overall lab health snapshot collected."
)
answer = await ask_llm(agent_id, message, context, log=log)
if not answer:
await log("warn", "llm", "LLM fallback — returning raw context")
answer = fallback_answer(agent_id, context)
answer = " ".join(answer_parts)
await asyncio.sleep(0.6)
if not approval_created:
proposed = detect_agent_proposed_action(answer, message)
if proposed:
with SessionLocal() as db:
await create_approval_request(
db=db,
ApprovalModel=Approval,
agent_id=agent_id,
action=proposed["action"],
reason=proposed["reason"],
action_type=proposed["action_type"],
target=proposed.get("target", ""),
terminal_log=terminal_log,
mirror_supervisors=mirror_to_supervisors,
publish=publish_event,
add_feed=add_feed,
)
approval_created = True
answer = (
f"{answer}\n\n⏸ **Approval required** — this action is in the Approval Inbox. "
f"Mo & Bart have been notified and must approve before we execute."
)
await asyncio.sleep(0.3)
await terminal_log(agent_id, f"[{prompt_id}] ✓ Mission complete", level="ok", phase="done", prompt_id=prompt_id)
await publish_event({"type": "agent_return", "agent_id": agent_id, "zone": zone, "prompt_id": prompt_id})
feed = add_feed(agent_id, f"Prompt answered: {message[:80]}", "info")
feed = add_feed(agent_id, f"{agent_name} completed a response (see Comms)", "info")
await publish_event({"type": "feed", "entry": feed})
await publish_event({"type": "prompt_result", "prompt_id": prompt_id, "agent_id": agent_id, "answer": answer})
return answer
@@ -259,7 +669,9 @@ async def heartbeat_loop() -> None:
while True:
try:
status = await collect_status()
workload = await collect_workload()
await publish_event({"type": "status", "data": status})
await publish_event({"type": "workload", "data": workload})
for domain, info in status["domains"].items():
if info["level"] == "down":
agent = "data-custodian" if domain == "databases" else "infra-sentinel"
@@ -274,6 +686,15 @@ async def heartbeat_loop() -> None:
async def lifespan(app: FastAPI):
global redis_client
redis_client = aioredis.from_url(REDIS_URL, decode_responses=True)
set_terminal_publisher(publish_event)
init_terminals([a["id"] for a in AGENTS] + NODE_IDS)
for a in AGENTS:
await terminal_log(a["id"], f"{a['name']} terminal online — awaiting missions", level="info", phase="boot")
for nid in NODE_IDS:
if nid not in NODE_REGISTRY:
continue
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())
add_feed("infra-sentinel", "ATC Command Center API online", "info")
yield
@@ -283,6 +704,7 @@ async def lifespan(app: FastAPI):
app = FastAPI(title="ATC Command Center API", lifespan=lifespan)
app.include_router(storage_s3_router)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
@@ -294,7 +716,103 @@ app.add_middleware(
@app.get("/api/health")
async def health():
return {"ok": True, "ts": datetime.now(timezone.utc).isoformat()}
llm_ok = False
try:
async with httpx.AsyncClient(timeout=4.0) as client:
r = await client.get(f"{LLM_URL.rstrip('/')}/models", headers={"Authorization": f"Bearer {LLM_API_KEY}"})
llm_ok = r.status_code == 200
except Exception:
pass
return {
"ok": True,
"ts": datetime.now(timezone.utc).isoformat(),
"llm_url": LLM_URL,
"llm_ok": llm_ok,
"llm_model": LLM_MODEL,
"database": db_health(),
"db_init": _db_info,
}
async def collect_workload(*, fast: bool = True, use_cache: bool = True) -> dict[str, Any]:
import time as _time
now = _time.time()
if use_cache and _workload_cache.get("data") and now - float(_workload_cache.get("ts") or 0) < WORKLOAD_CACHE_TTL:
return _workload_cache["data"]
gpu = await collect_gpu()
snap = await collect_full_lab_context(gpu_data=gpu, include_inventory=not fast)
payload = build_workload_payload(snap)
_workload_cache["ts"] = now
_workload_cache["data"] = payload
return payload
async def get_presentation_data(*, use_cache: bool = True) -> dict[str, Any]:
import time as _time
now = _time.time()
if use_cache and _presentation_cache.get("data") and now - float(_presentation_cache.get("ts") or 0) < PRESENTATION_CACHE_TTL:
return _presentation_cache["data"]
gpu = await collect_gpu()
snap = await collect_full_lab_context(gpu_data=gpu, include_inventory=False)
data = build_presentation_payload(snap)
data["source"] = "live"
_presentation_cache["ts"] = now
_presentation_cache["data"] = data
return data
@app.get("/api/presentation")
async def get_presentation():
return await get_presentation_data()
@app.get("/api/presentation/html")
async def get_presentation_html():
from fastapi.responses import HTMLResponse
payload = await get_presentation_data()
return HTMLResponse(render_presentation_html(payload))
@app.get("/api/presentation/decks")
async def get_presentation_decks():
return {"live": True, "builtin": list_static_decks(), "uploaded": list_decks()}
@app.get("/api/presentation/decks/{deck_id}")
async def get_presentation_deck(deck_id: str):
if deck_id == "live":
return await get_presentation_data()
deck = get_static_deck(deck_id) or get_deck(deck_id)
if not deck:
return {"error": "deck not found"}
return deck
@app.get("/api/presentation/decks/{deck_id}/html")
async def get_presentation_deck_html(deck_id: str):
from fastapi.responses import HTMLResponse
if deck_id == "live":
payload = await get_presentation_data()
else:
payload = get_static_deck(deck_id) or get_deck(deck_id)
if not payload:
return HTMLResponse("<h1>Deck not found</h1>", status_code=404)
return HTMLResponse(render_presentation_html(payload))
@app.post("/api/presentation/upload")
async def upload_presentation(file: UploadFile = File(...)):
content = await file.read()
if len(content) > 50 * 1024 * 1024:
return {"error": "file too large (max 50MB)"}
deck = await save_upload(file.filename or "upload.pptx", content)
return {"ok": True, "deck": deck}
@app.get("/api/workload")
async def get_workload(fast: bool = True):
return await collect_workload(fast=fast, use_cache=True)
@app.get("/api/status")
@@ -302,9 +820,95 @@ async def get_status():
return await collect_status()
@app.get("/api/gpu")
async def get_gpu():
return await collect_gpu()
def agent_stats() -> dict[str, dict[str, Any]]:
stats: dict[str, dict[str, Any]] = {a["id"]: {"tasks": 0, "last_active": None, "alerts": 0} for a in AGENTS}
with SessionLocal() as db:
rows = db.execute(select(FeedEntry).order_by(FeedEntry.ts.desc()).limit(200)).scalars().all()
for r in rows:
aid = r.agent_id
if aid not in stats:
continue
stats[aid]["tasks"] += 1
if r.level == "warn":
stats[aid]["alerts"] += 1
if stats[aid]["last_active"] is None and r.ts:
stats[aid]["last_active"] = r.ts.isoformat()
return stats
@app.get("/api/agents")
async def get_agents():
return {"agents": AGENTS, "zones": ZONES}
stats = agent_stats()
enriched = [{**a, "stats": stats.get(a["id"], {})} for a in AGENTS]
return {"agents": enriched, "zones": ZONES}
@app.get("/api/terminals")
async def get_terminals(limit: int = 200):
return {"terminals": get_all_terminals(limit)}
@app.get("/api/terminals/{subject_id}")
async def get_subject_terminal(subject_id: str, limit: int = 200):
valid_agents = {a["id"] for a in AGENTS}
if subject_id not in valid_agents and not is_node_id(subject_id):
return {"error": "unknown subject"}
return {"agent_id": subject_id, "lines": get_terminal_lines(subject_id, limit)}
@app.get("/api/nodes")
async def list_nodes():
workload = await collect_workload()
nodes = workload.get("topology", {}).get("nodes", [])
return {"nodes": [{"id": n["id"], "label": n["label"], "ip": n["ip"], "level": n["level"]} for n in nodes]}
@app.get("/api/nodes/{node_id}")
async def get_node(node_id: str):
if not is_node_id(node_id):
return {"error": "unknown node"}
workload = await collect_workload()
wn = next((n for n in workload.get("topology", {}).get("nodes", []) if n["id"] == node_id), None)
gpu = await collect_gpu()
snap = await collect_full_lab_context(gpu_data=gpu)
return build_node_detail(node_id, snap, wn)
@app.post("/api/nodes/{node_id}/probe")
async def post_node_probe(node_id: str):
if not is_node_id(node_id):
return {"error": "unknown node"}
asyncio.create_task(run_node_probe_task(node_id))
return {"ok": True, "node_id": node_id, "status": "probing"}
async def run_node_ask_task(node_id: str, message: str) -> None:
agent_id = NODE_AGENT.get(node_id, "infra-sentinel")
meta = NODE_REGISTRY[node_id]
await terminal_log(node_id, f"▶ Query: {message}", level="info", phase="ask")
await terminal_log(node_id, f"→ Routing to agent {agent_id}", level="info", phase="ask")
log = make_logger(node_id)
status = await collect_status()
context = await gather_agent_context(agent_id, status, log=log)
node_ctx = f"\n\n=== FOCUSED NODE: {meta['label']} ({meta['ip']}) ===\n{meta.get('description', '')}\n"
answer = await ask_llm(agent_id, message, context + node_ctx, log=log)
if not answer:
answer = fallback_answer(agent_id, context)
await terminal_log(node_id, f"◆ {answer}", level="llm", phase="answer")
await publish_event({"type": "node_ask_result", "node_id": node_id, "agent_id": agent_id, "answer": answer})
@app.post("/api/nodes/{node_id}/ask")
async def post_node_ask(node_id: str, body: NodeAskRequest):
if not is_node_id(node_id):
return {"error": "unknown node"}
asyncio.create_task(run_node_ask_task(node_id, body.message))
return {"ok": True, "node_id": node_id, "agent_id": NODE_AGENT.get(node_id), "status": "processing"}
@app.get("/api/feed")
@@ -326,47 +930,73 @@ async def get_feed(limit: int = 50):
@app.get("/api/approvals")
async def get_approvals():
async def get_approvals(status: str = "pending", limit: int = 100):
with SessionLocal() as db:
rows = db.execute(select(Approval).where(Approval.status == "pending")).scalars().all()
return {
"approvals": [
{
"id": r.id,
"ts": r.ts.isoformat() if r.ts else None,
"agent_id": r.agent_id,
"action": r.action,
"reason": r.reason,
"status": r.status,
}
for r in rows
]
}
items = list_approvals(db, Approval, status=status, limit=limit)
stats = approval_stats(db, Approval)
return {"approvals": items, "stats": stats, "action_types": APPROVAL_ACTION_TYPES}
@app.get("/api/approvals/stats")
async def get_approval_stats():
with SessionLocal() as db:
return approval_stats(db, Approval)
@app.post("/api/approvals")
async def post_approval(body: ApprovalCreateRequest):
valid_ids = {a["id"] for a in AGENTS}
if body.agent_id not in valid_ids:
return {"error": "unknown agent_id"}
if body.action_type not in APPROVAL_ACTION_TYPES:
body.action_type = "generic.mutate"
with SessionLocal() as db:
item = await create_approval_request(
db=db,
ApprovalModel=Approval,
agent_id=body.agent_id,
action=body.action,
reason=body.reason,
action_type=body.action_type,
target=body.target,
payload=body.payload,
priority=body.priority,
terminal_log=terminal_log,
mirror_supervisors=mirror_to_supervisors,
publish=publish_event,
add_feed=add_feed,
)
return {"ok": True, "approval": item}
@app.post("/api/approvals/{approval_id}/decide")
async def decide_approval(approval_id: str, body: ApprovalDecision):
valid_supervisors = {"mo-commander", "bart-commander"}
decided_by = body.decided_by if body.decided_by in valid_supervisors else "mo-commander"
with SessionLocal() as db:
row = db.get(Approval, approval_id)
if not row:
return {"error": "not found"}
row.status = "approved" if body.approved else "denied"
db.commit()
agent_id = row.agent_id
action = row.action
msg = f"Approval {'approved' if body.approved else 'denied'}: {action}"
feed = add_feed(agent_id, msg, "info" if body.approved else "warn")
await publish_event({"type": "feed", "entry": feed})
await publish_event({"type": "approval_update", "id": approval_id, "status": row.status})
return {"ok": True, "status": row.status}
item = await decide_approval_request(
db=db,
ApprovalModel=Approval,
approval_id=approval_id,
approved=body.approved,
decided_by=decided_by,
note=body.note,
terminal_log=terminal_log,
publish=publish_event,
add_feed=add_feed,
)
if not item:
return {"error": "not found"}
return {"ok": True, "approval": item}
@app.post("/api/prompt")
async def post_prompt(body: PromptRequest):
prompt_id = str(uuid.uuid4())[:8]
agent_id = route_agent(body.message)
valid_ids = {a["id"] for a in AGENTS}
agent_id = body.agent_id if body.agent_id in valid_ids else route_agent(body.message)
add_feed(agent_id, f"Prompt received: {body.message}", "info")
asyncio.create_task(run_agent_task(agent_id, body.message, prompt_id))
asyncio.create_task(_run_agent_task_safe(agent_id, body.message, prompt_id))
return {"prompt_id": prompt_id, "agent_id": agent_id, "status": "dispatched"}
@@ -376,7 +1006,13 @@ async def ws_ops(websocket: WebSocket):
ws_clients.add(websocket)
try:
status = await collect_status()
workload = await collect_workload()
await websocket.send_text(json.dumps({"type": "status", "data": status}, default=str))
await websocket.send_text(json.dumps({"type": "workload", "data": workload}, default=str))
await websocket.send_text(json.dumps({
"type": "terminal_history",
"terminals": get_all_terminals(150),
}, default=str))
while True:
await websocket.receive_text()
except WebSocketDisconnect: