From 7da2dd60f05340b455c854dafacba03dddf4b344 Mon Sep 17 00:00:00 2001 From: mo Date: Thu, 25 Jun 2026 01:33:48 +0000 Subject: [PATCH] Add MySQL/MongoDB consoles, fit topology without scroll, longer flow lines. Compact workbench panel, collapsible infra bar, narrower topology nodes with wider inter-stage gaps for visible connection lines. --- api/sql_console.py | 248 ++++++++++++++---- ui/src/App.tsx | 40 ++- ui/src/components/features/AgentFleet.tsx | 8 +- ui/src/components/features/AgentWorkbench.tsx | 62 ++--- .../components/features/PlatformTopology.tsx | 20 +- ui/src/components/features/SqlWorkbench.tsx | 93 ++++--- ui/src/components/features/WorkbenchPanel.tsx | 12 +- ui/src/hooks/useCommandCenter.ts | 12 +- ui/src/styles/globals.css | 10 +- 9 files changed, 343 insertions(+), 162 deletions(-) diff --git a/api/sql_console.py b/api/sql_console.py index fd841a9..eae1e20 100644 --- a/api/sql_console.py +++ b/api/sql_console.py @@ -1,23 +1,35 @@ -"""Live SQL console — PostgreSQL + Trino with benchmark.""" +"""Live SQL console — PostgreSQL, MySQL, MongoDB + Trino with benchmark.""" from __future__ import annotations import os +import re import time from typing import Any import httpx import psycopg2 +import pymysql from fastapi import APIRouter from fastapi.responses import JSONResponse from pydantic import BaseModel, Field +from pymongo import MongoClient -PG_HOST = os.getenv("DB_VAULT_HOST", "10.0.21.51") +DB_HOST = os.getenv("DB_VAULT_HOST", "10.0.21.51") PG_PORT = int(os.getenv("PG_PORT", "5432")) PG_USER = os.getenv("PG_USER", "mo") PG_PASS = os.getenv("PG_PASSWORD", "Dell2026!") PG_DB = os.getenv("PG_DATABASE", "postgres") +MYSQL_PORT = int(os.getenv("MYSQL_PORT", "3306")) +MYSQL_USER = os.getenv("MYSQL_USER", "mo") +MYSQL_PASS = os.getenv("MYSQL_PASSWORD", "Dell2026!") +MYSQL_DB = os.getenv("MYSQL_DATABASE", "hr") + +MONGO_HOST = os.getenv("MONGO_HOST", DB_HOST) +MONGO_PORT = int(os.getenv("MONGO_PORT", "27017")) +MONGO_DB = os.getenv("MONGO_DATABASE", "supplychain") + TRINO_URL = os.getenv("TRINO_URL", "http://10.0.21.50:8089").rstrip("/") TRINO_USER = os.getenv("TRINO_USER", "atc") @@ -36,6 +48,30 @@ SAMPLES: dict[str, list[dict[str, str]]] = { {"id": "pg9", "label": "Explain scan 100k", "sql": "EXPLAIN ANALYZE SELECT count(*) FROM generate_series(1, 100000);"}, {"id": "pg10", "label": "Memory settings", "sql": "SELECT name, setting, unit FROM pg_settings WHERE name IN ('max_connections','shared_buffers','work_mem','effective_cache_size');"}, ], + "mysql": [ + {"id": "my1", "label": "Server version", "sql": "SELECT VERSION() AS version;"}, + {"id": "my2", "label": "Current session", "sql": "SELECT DATABASE() AS db, USER() AS user, NOW() AS now_ts;"}, + {"id": "my3", "label": "HR tables", "sql": "SHOW TABLES;"}, + {"id": "my4", "label": "Table sizes", "sql": "SELECT table_name, table_rows, ROUND((data_length+index_length)/1024/1024,1) AS mb FROM information_schema.tables WHERE table_schema='hr' ORDER BY (data_length+index_length) DESC;"}, + {"id": "my5", "label": "Employee events sample", "sql": "SELECT event_type, COUNT(*) AS cnt FROM employee_events GROUP BY event_type ORDER BY cnt DESC LIMIT 10;"}, + {"id": "my6", "label": "Recent events", "sql": "SELECT employee_id, event_type, event_date FROM employee_events ORDER BY event_date DESC LIMIT 10;"}, + {"id": "my7", "label": "Active connections", "sql": "SHOW STATUS LIKE 'Threads_connected';"}, + {"id": "my8", "label": "InnoDB buffer pool", "sql": "SHOW STATUS LIKE 'Innodb_buffer_pool%';"}, + {"id": "my9", "label": "Explain count scan", "sql": "EXPLAIN SELECT COUNT(*) FROM employee_events;"}, + {"id": "my10", "label": "Departments overview", "sql": "SELECT department, COUNT(*) AS employees FROM employees GROUP BY department ORDER BY employees DESC LIMIT 10;"}, + ], + "mongodb": [ + {"id": "mg1", "label": "List databases", "sql": "SHOW DATABASES"}, + {"id": "mg2", "label": "Supplychain collections", "sql": "SHOW COLLECTIONS supplychain"}, + {"id": "mg3", "label": "Count events", "sql": "COUNT supplychain.events"}, + {"id": "mg4", "label": "Sample document", "sql": "FIND supplychain.events LIMIT 3"}, + {"id": "mg5", "label": "Events by type", "sql": "AGGREGATE supplychain.events GROUP type TOP 10"}, + {"id": "mg6", "label": "Events by region", "sql": "AGGREGATE supplychain.events GROUP region TOP 10"}, + {"id": "mg7", "label": "Distinct event types", "sql": "DISTINCT supplychain.events type"}, + {"id": "mg8", "label": "Server status", "sql": "SERVER STATUS"}, + {"id": "mg9", "label": "Indexes on events", "sql": "INDEXES supplychain.events"}, + {"id": "mg10", "label": "Recent events", "sql": "FIND supplychain.events SORT timestamp DESC LIMIT 5"}, + ], "trino": [ {"id": "tq1", "label": "Trino version", "sql": "SELECT version()"}, {"id": "tq2", "label": "Catalogs", "sql": "SHOW CATALOGS"}, @@ -56,10 +92,22 @@ BENCHMARK_SQL = { } +def _tabular(columns: list[str], rows: list[list[Any]], elapsed_ms: int, **extra: Any) -> dict[str, Any]: + return { + "ok": True, + "columns": columns, + "rows": rows, + "row_count": len(rows), + "elapsed_ms": elapsed_ms, + "truncated": extra.pop("truncated", False), + **extra, + } + + def _run_postgres(sql: str, limit: int = 200) -> dict[str, Any]: t0 = time.perf_counter() conn = psycopg2.connect( - host=PG_HOST, port=PG_PORT, user=PG_USER, password=PG_PASS, dbname=PG_DB, connect_timeout=8, + host=DB_HOST, port=PG_PORT, user=PG_USER, password=PG_PASS, dbname=PG_DB, connect_timeout=8, ) try: conn.set_session(readonly=True, autocommit=True) @@ -68,14 +116,112 @@ def _run_postgres(sql: str, limit: int = 200) -> dict[str, Any]: elapsed_ms = int((time.perf_counter() - t0) * 1000) if cur.description: columns = [d[0] for d in cur.description] - rows = cur.fetchmany(limit) - data = [list(r) for r in rows] - return {"ok": True, "columns": columns, "rows": data, "row_count": len(data), "elapsed_ms": elapsed_ms, "truncated": len(data) >= limit} - return {"ok": True, "columns": [], "rows": [], "row_count": 0, "elapsed_ms": elapsed_ms, "message": "OK"} + rows = [list(r) for r in cur.fetchmany(limit)] + return _tabular(columns, rows, elapsed_ms, truncated=len(rows) >= limit) + return _tabular([], [], elapsed_ms, message="OK") finally: conn.close() +def _run_mysql(sql: str, limit: int = 200) -> dict[str, Any]: + t0 = time.perf_counter() + conn = pymysql.connect( + host=DB_HOST, port=MYSQL_PORT, user=MYSQL_USER, password=MYSQL_PASS, + database=MYSQL_DB, connect_timeout=8, read_timeout=30, + ) + try: + cur = conn.cursor() + cur.execute(sql) + elapsed_ms = int((time.perf_counter() - t0) * 1000) + if cur.description: + columns = [d[0] for d in cur.description] + rows = [list(r) for r in cur.fetchmany(limit)] + return _tabular(columns, rows, elapsed_ms, truncated=len(rows) >= limit) + return _tabular([], [], elapsed_ms, message="OK") + finally: + conn.close() + + +def _mongo_client() -> MongoClient: + return MongoClient(f"mongodb://{MONGO_HOST}:{MONGO_PORT}/", serverSelectionTimeoutMS=8000) + + +def _fmt_mongo_value(val: Any) -> Any: + if val is None: + return None + if isinstance(val, (str, int, float, bool)): + return val + return str(val)[:200] + + +def _run_mongo(query: str, limit: int = 200) -> dict[str, Any]: + t0 = time.perf_counter() + q = query.strip() + client = _mongo_client() + try: + columns: list[str] = [] + rows: list[list[Any]] = [] + + if q == "SHOW DATABASES": + columns = ["database"] + rows = [[name] for name in client.list_database_names()] + elif m := re.match(r"^SHOW COLLECTIONS(?:\s+(\w+))?$", q, re.I): + db_name = m.group(1) or MONGO_DB + columns = ["collection"] + rows = [[name] for name in client[db_name].list_collection_names()] + elif m := re.match(r"^COUNT\s+(\w+)\.(\w+)$", q, re.I): + db_name, coll = m.group(1), m.group(2) + columns = ["count"] + rows = [[client[db_name][coll].estimated_document_count()]] + elif m := re.match(r"^FIND\s+(\w+)\.(\w+)(?:\s+SORT\s+(\w+)\s+DESC)?(?:\s+LIMIT\s+(\d+))?$", q, re.I): + db_name, coll, sort_field, lim = m.group(1), m.group(2), m.group(3), m.group(4) + n = min(int(lim) if lim else 10, limit) + cursor = client[db_name][coll].find({}) + if sort_field: + cursor = cursor.sort(sort_field, -1) + docs = list(cursor.limit(n)) + if not docs: + columns = ["result"] + rows = [["(empty)"]] + else: + columns = sorted({k for d in docs for k in d if k != "_id"}) + rows = [[_fmt_mongo_value(d.get(c)) for c in columns] for d in docs] + elif m := re.match(r"^DISTINCT\s+(\w+)\.(\w+)\s+(\w+)$", q, re.I): + db_name, coll, field = m.group(1), m.group(2), m.group(3) + columns = [field] + vals = client[db_name][coll].distinct(field)[:limit] + rows = [[_fmt_mongo_value(v)] for v in vals] + elif m := re.match(r"^AGGREGATE\s+(\w+)\.(\w+)\s+GROUP\s+(\w+)\s+TOP\s+(\d+)$", q, re.I): + db_name, coll, field, top_n = m.group(1), m.group(2), m.group(3), int(m.group(4)) + pipe = [ + {"$group": {"_id": f"${field}", "count": {"$sum": 1}}}, + {"$sort": {"count": -1}}, + {"$limit": min(top_n, limit)}, + ] + columns = [field, "count"] + rows = [[r.get("_id"), r.get("count")] for r in client[db_name][coll].aggregate(pipe, maxTimeMS=15000)] + elif m := re.match(r"^INDEXES\s+(\w+)\.(\w+)$", q, re.I): + db_name, coll = m.group(1), m.group(2) + columns = ["name", "keys"] + rows = [[i.get("name"), str(i.get("key"))] for i in client[db_name][coll].list_indexes()] + elif q.upper() == "SERVER STATUS": + status = client.admin.command("serverStatus") + columns = ["metric", "value"] + rows = [ + ["version", status.get("version")], + ["uptime_secs", status.get("uptime")], + ["connections_current", status.get("connections", {}).get("current")], + ["mem_resident_mb", status.get("mem", {}).get("resident")], + ] + else: + return {"ok": False, "error": f"Unknown MongoDB command: {q[:120]}"} + + elapsed_ms = int((time.perf_counter() - t0) * 1000) + return _tabular(columns, rows[:limit], elapsed_ms, truncated=len(rows) > limit) + finally: + client.close() + + def _run_trino(sql: str, limit: int = 200) -> dict[str, Any]: t0 = time.perf_counter() headers = {"X-Trino-User": TRINO_USER, "Content-Type": "text/plain"} @@ -104,19 +250,14 @@ def _run_trino(sql: str, limit: int = 200) -> dict[str, Any]: data = client.get(nxt, headers=headers).json() elapsed_ms = int((time.perf_counter() - t0) * 1000) stats = data.get("stats") or {} - return { - "ok": True, - "columns": columns, - "rows": rows, - "row_count": len(rows), - "elapsed_ms": elapsed_ms, - "engine_stats_ms": stats.get("elapsedTimeMillis"), - "truncated": len(rows) >= limit, - } + return _tabular(columns, rows, elapsed_ms, engine_stats_ms=stats.get("elapsedTimeMillis"), truncated=len(rows) >= limit) + + +ENGINES = ("postgres", "mysql", "mongodb", "trino") class SqlRequest(BaseModel): - engine: str = Field(..., pattern="^(postgres|trino)$") + engine: str = Field(..., pattern="^(postgres|mysql|mongodb|trino)$") sql: str = Field(..., min_length=1, max_length=8000) @@ -124,51 +265,58 @@ class SqlRequest(BaseModel): async def get_samples(engine: str): if engine not in SAMPLES: return JSONResponse({"error": "unknown engine"}, status_code=404) - return { - "engine": engine, - "samples": SAMPLES[engine], - "connection": _connection_info(engine), - } + return {"engine": engine, "samples": SAMPLES[engine], "connection": _connection_info(engine)} @router.get("/health") async def sql_health(): - pg_ok = trino_ok = False - pg_err = trino_err = None - try: - _run_postgres("SELECT 1") - pg_ok = True - except Exception as exc: - pg_err = str(exc)[:200] - try: - r = _run_trino("SELECT 1") - trino_ok = bool(r.get("ok")) - if not trino_ok: - trino_err = r.get("error") - except Exception as exc: - trino_err = str(exc)[:200] - return { - "postgres": {"ok": pg_ok, "host": PG_HOST, "user": PG_USER, "error": pg_err}, - "trino": {"ok": trino_ok, "url": TRINO_URL, "user": TRINO_USER, "error": trino_err}, - } + out: dict[str, Any] = {} + for eng in ENGINES: + try: + if eng == "postgres": + _run_postgres("SELECT 1") + out[eng] = {"ok": True, "host": DB_HOST, "user": PG_USER, "error": None} + elif eng == "mysql": + _run_mysql("SELECT 1") + out[eng] = {"ok": True, "host": DB_HOST, "database": MYSQL_DB, "user": MYSQL_USER, "error": None} + elif eng == "mongodb": + _run_mongo("SHOW DATABASES") + out[eng] = {"ok": True, "host": MONGO_HOST, "database": MONGO_DB, "error": None} + else: + r = _run_trino("SELECT 1") + out[eng] = {"ok": bool(r.get("ok")), "url": TRINO_URL, "user": TRINO_USER, "error": r.get("error")} + except Exception as exc: + out[eng] = {"ok": False, "error": str(exc)[:200]} + return out def _connection_info(engine: str) -> dict[str, str]: if engine == "postgres": - return {"host": PG_HOST, "port": str(PG_PORT), "database": PG_DB, "user": PG_USER} + return {"host": DB_HOST, "port": str(PG_PORT), "database": PG_DB, "user": PG_USER} + if engine == "mysql": + return {"host": DB_HOST, "port": str(MYSQL_PORT), "database": MYSQL_DB, "user": MYSQL_USER} + if engine == "mongodb": + return {"host": MONGO_HOST, "port": str(MONGO_PORT), "database": MONGO_DB} return {"url": TRINO_URL, "user": TRINO_USER} +def _dispatch(engine: str, sql: str) -> dict[str, Any]: + if engine == "postgres": + return _run_postgres(sql) + if engine == "mysql": + return _run_mysql(sql) + if engine == "mongodb": + return _run_mongo(sql) + return _run_trino(sql) + + @router.post("/execute") async def execute_sql(body: SqlRequest): sql = body.sql.strip().rstrip(";") if not sql: return JSONResponse({"ok": False, "error": "Empty query"}, status_code=400) try: - if body.engine == "postgres": - result = _run_postgres(sql) - else: - result = _run_trino(sql) + result = _dispatch(body.engine, sql) if not result.get("ok"): return JSONResponse(result, status_code=422) return {**result, "engine": body.engine, "sql": sql} @@ -182,21 +330,17 @@ async def benchmark(): results: dict[str, Any] = {} for engine, sql in BENCHMARK_SQL.items(): try: - if engine == "postgres": - results[engine] = _run_postgres(sql) - else: - results[engine] = _run_trino(sql) + results[engine] = _dispatch(engine, sql) results[engine]["sql"] = sql except Exception as exc: results[engine] = {"ok": False, "error": str(exc)[:300], "sql": sql} pg_ms = results.get("postgres", {}).get("elapsed_ms") tr_ms = results.get("trino", {}).get("elapsed_ms") faster = None + speedup = None if pg_ms and tr_ms: faster = "trino" if tr_ms < pg_ms else "postgres" - speedup = round(pg_ms / tr_ms, 2) if tr_ms and tr_ms < pg_ms else round(tr_ms / pg_ms, 2) if pg_ms else None - else: - speedup = None + speedup = round(pg_ms / tr_ms, 2) if tr_ms < pg_ms else round(tr_ms / pg_ms, 2) return { "ok": True, "postgres": results.get("postgres"), diff --git a/ui/src/App.tsx b/ui/src/App.tsx index 29d3e12..35c2ffb 100644 --- a/ui/src/App.tsx +++ b/ui/src/App.tsx @@ -1,4 +1,5 @@ import { useRef, useState } from 'react' +import { ChevronDown, ChevronUp } from 'lucide-react' import { useClock } from './hooks/useClock' import { useCommandCenter } from './hooks/useCommandCenter' import { useLiveMetrics } from './hooks/useLiveMetrics' @@ -25,6 +26,7 @@ export default function App() { const clock = useClock() const cc = useCommandCenter() const [gpuChatActive, setGpuChatActive] = useState(false) + const [infraOpen, setInfraOpen] = useState(false) const gpuBoost = gpuChatActive || cc.mainView === 'knowledge' const { agentLoads, gpuLive } = useLiveMetrics(cc.agents, cc.gpu, cc.anims, gpuBoost) const mainScrollRef = useRef(null) @@ -74,9 +76,9 @@ export default function App() { onSelectZone={cc.selectNode} /> -
+
-
+
{isPlatform && ( <>
@@ -91,20 +93,30 @@ export default function App() { />
- + + {infraOpen && ( + + )} )} -
+
{cc.mainView === 'platform' ? ( {isPlatform && ( -
+
- - + +
-

{agent.name.split(' ·')[0]}

+

{agent.name.split(' ·')[0]}

{meta.domain}

diff --git a/ui/src/components/features/AgentWorkbench.tsx b/ui/src/components/features/AgentWorkbench.tsx index 2d33dd5..cbf7b02 100644 --- a/ui/src/components/features/AgentWorkbench.tsx +++ b/ui/src/components/features/AgentWorkbench.tsx @@ -10,6 +10,7 @@ type Props = { lines: TerminalLine[] busy: boolean onSendPrompt: (message: string, agentId?: string) => void + compact?: boolean } const LEVEL: Record = { @@ -23,7 +24,7 @@ const LEVEL: Record = { probe: 'text-amber-300', } -export function AgentWorkbench({ agent, lines, busy, onSendPrompt }: Props) { +export function AgentWorkbench({ agent, lines, busy, onSendPrompt, compact }: Props) { const meta = getAgentMeta(agent.id) const Icon = meta.icon const bottomRef = useRef(null) @@ -33,49 +34,44 @@ export function AgentWorkbench({ agent, lines, busy, onSendPrompt }: Props) { }, [lines, busy]) return ( -
-
+
+
- +
-

{agent.name}

-

{meta.domain} · {agent.zone}

-

{agent.role}

+

{agent.name}

+ {!compact &&

{meta.domain} · {agent.zone}

}
-
- {busy && ( - - Live - - )} - - Agent terminal +
+ {busy && } + + Terminal
-
- {(agent.suggested_prompts || []).slice(0, 6).map((prompt) => ( - - ))} -
+ {!compact && ( +
+ {(agent.suggested_prompts || []).slice(0, 4).map((prompt) => ( + + ))} +
+ )} -
+
{lines.length === 0 && ( -

- Agent terminal ready — probe output and LLM responses appear here -

+

Agent terminal ready

)} {lines.map((line) => (
diff --git a/ui/src/components/features/PlatformTopology.tsx b/ui/src/components/features/PlatformTopology.tsx index c55a1a1..5e8c032 100644 --- a/ui/src/components/features/PlatformTopology.tsx +++ b/ui/src/components/features/PlatformTopology.tsx @@ -153,13 +153,17 @@ STAGES.forEach((stage, col) => { }) function nodeCoords(col: number, row: number, rows: number) { - const colW = 100 / 5 - const yPad = 8 - const ySpan = 84 + const colCount = 5 + const gap = 3.2 + const colW = (100 - gap * (colCount - 1)) / colCount + const xCenter = col * (colW + gap) + colW / 2 + const nodeHalf = colW * 0.34 + const yPad = 5 + const ySpan = 90 const y = yPad + ((row + 0.5) / rows) * ySpan return { - inX: col * colW + colW * 0.08, - outX: col * colW + colW * 0.92, + inX: xCenter - nodeHalf, + outX: xCenter + nodeHalf, y, } } @@ -279,7 +283,7 @@ export function PlatformTopology({ workload, animations, selectedNodeId, onNodeC return (
@@ -290,7 +294,7 @@ export function PlatformTopology({ workload, animations, selectedNodeId, onNodeC

Data Platform Topology

- Click PostgreSQL or Trino → live SQL console · agents → terminal below + Click PostgreSQL / MySQL / MongoDB / Trino → live console · agents → terminal below

@@ -361,7 +365,7 @@ export function PlatformTopology({ workload, animations, selectedNodeId, onNodeC })} -
+
{STAGES.map((stage) => (
diff --git a/ui/src/components/features/SqlWorkbench.tsx b/ui/src/components/features/SqlWorkbench.tsx index 0f56c20..606d96a 100644 --- a/ui/src/components/features/SqlWorkbench.tsx +++ b/ui/src/components/features/SqlWorkbench.tsx @@ -14,32 +14,47 @@ type SqlResult = { sql?: string } +type Engine = 'postgres' | 'mysql' | 'mongodb' | 'trino' + type Props = { - engine: 'postgres' | 'trino' + engine: Engine + compact?: boolean } -const ENGINE_META = { +const ENGINE_META: Record = { postgres: { title: 'PostgreSQL SQL Console', sub: 'DB Vault · 10.0.21.51:5432 · user mo', accent: 'text-blue-400', + queryLabel: 'SQL', + }, + mysql: { + title: 'MySQL SQL Console', + sub: 'DB Vault · 10.0.21.51:3306 · database hr · user mo', + accent: 'text-amber-400', + queryLabel: 'SQL', + }, + mongodb: { + title: 'MongoDB Query Console', + sub: 'DB Vault · 10.0.21.51:27017 · supplychain', + accent: 'text-emerald-400', + queryLabel: 'Command', }, trino: { title: 'Trino SQL Console', - sub: 'Lakehouse · 10.0.21.50:8089 · federated lakehouse queries', + sub: 'Lakehouse · 10.0.21.50:8089 · federated queries', accent: 'text-violet-400', + queryLabel: 'SQL', }, } -export function SqlWorkbench({ engine }: Props) { +export function SqlWorkbench({ engine, compact }: Props) { const meta = ENGINE_META[engine] const [samples, setSamples] = useState([]) const [sql, setSql] = useState('SELECT version();') const [result, setResult] = useState(null) const [benchmark, setBenchmark] = useState<{ comparison?: { postgres_ms?: number; trino_ms?: number; faster?: string; speedup_factor?: number } - postgres?: SqlResult - trino?: SqlResult } | null>(null) const [loading, setLoading] = useState(false) const [benchLoading, setBenchLoading] = useState(false) @@ -106,40 +121,41 @@ export function SqlWorkbench({ engine }: Props) { } return ( -
-
+
+
-

- +

+ {meta.title}

-

{meta.sub}

+

{meta.sub}

-
- - + {(engine === 'postgres' || engine === 'trino') && ( + + )}
-
-