From 170eb2418b94633d3616858876eeb73b6d4ae5e3 Mon Sep 17 00:00:00 2001 From: mo Date: Thu, 25 Jun 2026 01:10:57 +0000 Subject: [PATCH] Add agent workbench terminal and live PostgreSQL/Trino SQL consoles. Clicking agents opens a dedicated terminal panel; topology PostgreSQL/Trino nodes open SQL workbench with ten demo queries and Postgres vs Trino benchmark. --- api/Dockerfile | 2 +- api/main.py | 2 + api/sql_console.py | 211 ++++++++++++++++++ ui/src/App.tsx | 31 ++- ui/src/components/features/AgentFleet.tsx | 17 +- ui/src/components/features/AgentWorkbench.tsx | 93 ++++++++ .../components/features/PlatformTopology.tsx | 2 +- ui/src/components/features/SqlWorkbench.tsx | 205 +++++++++++++++++ ui/src/components/features/WorkbenchPanel.tsx | 25 +++ ui/src/hooks/useCommandCenter.ts | 29 ++- 10 files changed, 598 insertions(+), 19 deletions(-) create mode 100644 api/sql_console.py create mode 100644 ui/src/components/features/AgentWorkbench.tsx create mode 100644 ui/src/components/features/SqlWorkbench.tsx create mode 100644 ui/src/components/features/WorkbenchPanel.tsx diff --git a/api/Dockerfile b/api/Dockerfile index 0cf56f4..2bc9dad 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 . +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 . RUN mkdir -p /data ENV DATABASE_URL=sqlite:////data/atc-agents.db EXPOSE 3201 diff --git a/api/main.py b/api/main.py index 677fe2c..6b90a0e 100644 --- a/api/main.py +++ b/api/main.py @@ -29,6 +29,7 @@ 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 elasticsearch_api import router as elasticsearch_router +from sql_console import router as sql_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 ( @@ -707,6 +708,7 @@ async def lifespan(app: FastAPI): app = FastAPI(title="ATC Command Center API", lifespan=lifespan) app.include_router(storage_s3_router) app.include_router(elasticsearch_router) +app.include_router(sql_router) app.add_middleware( CORSMiddleware, allow_origins=["*"], diff --git a/api/sql_console.py b/api/sql_console.py new file mode 100644 index 0000000..335561b --- /dev/null +++ b/api/sql_console.py @@ -0,0 +1,211 @@ +"""Live SQL console — PostgreSQL + Trino with benchmark.""" + +from __future__ import annotations + +import os +import time +from typing import Any + +import httpx +import psycopg2 +from fastapi import APIRouter +from fastapi.responses import JSONResponse +from pydantic import BaseModel, Field + +PG_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") + +TRINO_URL = os.getenv("TRINO_URL", "http://10.0.21.50:8089").rstrip("/") +TRINO_USER = os.getenv("TRINO_USER", "atc") + +router = APIRouter(prefix="/api/sql", tags=["sql"]) + +SAMPLES: dict[str, list[dict[str, str]]] = { + "postgres": [ + {"id": "pg1", "label": "Server version", "sql": "SELECT version();"}, + {"id": "pg2", "label": "Current session", "sql": "SELECT current_database(), current_user, now();"}, + {"id": "pg3", "label": "Public tables", "sql": "SELECT tablename FROM pg_tables WHERE schemaname='public' ORDER BY 1 LIMIT 20;"}, + {"id": "pg4", "label": "Table count", "sql": "SELECT count(*) AS public_tables FROM information_schema.tables WHERE table_schema='public';"}, + {"id": "pg5", "label": "Active queries", "sql": "SELECT pid, usename, state, left(query, 100) AS query FROM pg_stat_activity WHERE state='active' AND pid <> pg_backend_pid() LIMIT 10;"}, + {"id": "pg6", "label": "Database sizes", "sql": "SELECT datname, pg_size_pretty(pg_database_size(datname)) AS size FROM pg_database ORDER BY pg_database_size(datname) DESC LIMIT 10;"}, + {"id": "pg7", "label": "Row counts (stats)", "sql": "SELECT relname, n_live_tup FROM pg_stat_user_tables ORDER BY n_live_tup DESC NULLS LAST LIMIT 10;"}, + {"id": "pg8", "label": "Connections", "sql": "SELECT count(*) AS connections, state FROM pg_stat_activity GROUP BY state;"}, + {"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');"}, + ], + "trino": [ + {"id": "tq1", "label": "Trino version", "sql": "SELECT version()"}, + {"id": "tq2", "label": "Catalogs", "sql": "SHOW CATALOGS"}, + {"id": "tq3", "label": "Iceberg schemas", "sql": "SHOW SCHEMAS FROM iceberg"}, + {"id": "tq4", "label": "Iceberg tables", "sql": "SHOW TABLES FROM iceberg.default"}, + {"id": "tq5", "label": "Federated PG catalog", "sql": "SHOW SCHEMAS FROM postgres_sales"}, + {"id": "tq6", "label": "Cluster nodes", "sql": "SELECT node_id, http_uri, state FROM system.runtime.nodes"}, + {"id": "tq7", "label": "Running queries", "sql": "SELECT query_id, state, user FROM system.runtime.queries WHERE state != 'FINISHED' LIMIT 10"}, + {"id": "tq8", "label": "Count 1M rows (fast)", "sql": "SELECT count(*) AS rows FROM range(1, 1000000)"}, + {"id": "tq9", "label": "Explain 5M scan", "sql": "EXPLAIN SELECT count(*) FROM range(1, 5000000)"}, + {"id": "tq10", "label": "Sum benchmark prep", "sql": "SELECT sum(x) AS total FROM (SELECT x FROM unnest(sequence(1, 500000)) t(x))"}, + ], +} + +BENCHMARK_SQL = { + "postgres": "SELECT count(*) AS rows FROM generate_series(1, 2000000)", + "trino": "SELECT count(*) AS rows FROM range(1, 2000000)", +} + + +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, + ) + try: + conn.set_session(readonly=True, autocommit=True) + 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 = 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"} + finally: + conn.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"} + with httpx.Client(timeout=120.0) as client: + resp = client.post(f"{TRINO_URL}/v1/statement", content=sql, headers=headers) + if resp.status_code >= 400: + return {"ok": False, "error": resp.text[:500]} + data = resp.json() + if data.get("error"): + return {"ok": False, "error": str(data["error"])[:500]} + columns: list[str] = [] + rows: list[list[Any]] = [] + while True: + if data.get("columns") and not columns: + columns = [c["name"] for c in data["columns"]] + if data.get("data"): + rows.extend(data["data"]) + if len(rows) >= limit: + rows = rows[:limit] + break + if data.get("error"): + return {"ok": False, "error": str(data["error"])[:500]} + nxt = data.get("nextUri") + if not nxt: + break + 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, + } + + +class SqlRequest(BaseModel): + engine: str = Field(..., pattern="^(postgres|trino)$") + sql: str = Field(..., min_length=1, max_length=8000) + + +@router.get("/samples/{engine}") +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), + } + + +@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}, + } + + +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 {"url": TRINO_URL, "user": TRINO_USER} + + +@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) + if not result.get("ok"): + return JSONResponse(result, status_code=422) + return {**result, "engine": body.engine, "sql": sql} + except Exception as exc: + return JSONResponse({"ok": False, "error": str(exc)[:500]}, status_code=502) + + +@router.post("/benchmark") +async def benchmark(): + """Run equivalent count(*) on Postgres vs Trino to demo lakehouse speed.""" + 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]["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 + 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 + return { + "ok": True, + "postgres": results.get("postgres"), + "trino": results.get("trino"), + "comparison": { + "postgres_ms": pg_ms, + "trino_ms": tr_ms, + "faster": faster, + "speedup_factor": speedup, + "note": "Same logical workload: count 2M rows — Trino distributed engine vs PostgreSQL OLTP", + }, + } diff --git a/ui/src/App.tsx b/ui/src/App.tsx index 9e87d0c..29d3e12 100644 --- a/ui/src/App.tsx +++ b/ui/src/App.tsx @@ -17,6 +17,7 @@ import { KnowledgeChatView } from './components/features/KnowledgeChatView' import { StorageView } from './components/features/StorageView' import { SearchView } from './components/features/SearchView' import { TerminalDock } from './components/features/TerminalDock' +import { WorkbenchPanel } from './components/features/WorkbenchPanel' import { resolveInfraNode } from './lib/infraCatalog' import { cn } from './lib/utils' @@ -103,7 +104,7 @@ export default function App() { )} -
+
{cc.mainView === 'platform' ? ( )}
+ + {isPlatform && cc.workbenchMode && ( + + )}
{isPlatform && ( @@ -147,14 +158,16 @@ export default function App() { onOpenTerminal={cc.openTerminal} onProbeNodeId={cc.probeNodeId} /> - cc.setTerminalExpanded(!cc.terminalExpanded)} - /> + {!cc.workbenchMode && ( + cc.setTerminalExpanded(!cc.terminalExpanded)} + /> + )} )} diff --git a/ui/src/components/features/AgentFleet.tsx b/ui/src/components/features/AgentFleet.tsx index 92fd5df..b6267ce 100644 --- a/ui/src/components/features/AgentFleet.tsx +++ b/ui/src/components/features/AgentFleet.tsx @@ -36,20 +36,25 @@ function AgentCard({ type="button" onClick={onSelect} className={cn( - 'flex h-[118px] w-[140px] shrink-0 flex-col gap-1 rounded-lg border bg-surface-raised p-1.5 text-left shadow-sm dark:bg-surface-overlay', - selected ? 'border-docker/50 ring-1 ring-docker/20' : 'border-border hover:border-border-strong', + 'flex h-[136px] w-[168px] shrink-0 flex-col gap-1 rounded-lg border bg-surface-raised p-1.5 text-left shadow-sm dark:bg-surface-overlay', + selected ? 'border-docker ring-2 ring-docker/30 shadow-[0_0_20px_rgba(13,183,237,0.15)]' : 'border-border hover:border-docker/30', )} >
- - + +
-

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

+

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

{meta.domain}

+ {selected && ( + + Terminal open + + )}

{agent.role}

{agentTaskLabel(agent.id, anim)}

@@ -75,7 +80,7 @@ export function AgentFleet({ agents, animations, selectedId, loads, approvalCoun

Agent Fleet

-

Klik agent → stel vraag in chat · elk agent bewaakt één domein

+

Click agent → dedicated terminal opens below

+ ))} +
+ +
+ {lines.length === 0 && ( +

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

+ )} + {lines.map((line) => ( +
+ + {line.ts ? new Date(line.ts).toLocaleTimeString('en-US', { hour12: false }) : ''} + {' '} + [{line.phase}] {line.text} +
+ ))} + {busy && █} +
+
+
+ ) +} diff --git a/ui/src/components/features/PlatformTopology.tsx b/ui/src/components/features/PlatformTopology.tsx index a967816..c55a1a1 100644 --- a/ui/src/components/features/PlatformTopology.tsx +++ b/ui/src/components/features/PlatformTopology.tsx @@ -290,7 +290,7 @@ export function PlatformTopology({ workload, animations, selectedNodeId, onNodeC

Data Platform Topology

- Airflow daily Python → CDC → stream → lakehouse → consumers + Click PostgreSQL or Trino → live SQL console · agents → terminal below

diff --git a/ui/src/components/features/SqlWorkbench.tsx b/ui/src/components/features/SqlWorkbench.tsx new file mode 100644 index 0000000..c63f182 --- /dev/null +++ b/ui/src/components/features/SqlWorkbench.tsx @@ -0,0 +1,205 @@ +import { useCallback, useEffect, useState } from 'react' +import { Database, Loader2, Play, Zap } from 'lucide-react' +import { cn } from '../../lib/utils' +import { subTabActive, subTabIdle } from '../../lib/tabActive' + +type Sample = { id: string; label: string; sql: string } +type SqlResult = { + ok: boolean + columns?: string[] + rows?: unknown[][] + row_count?: number + elapsed_ms?: number + error?: string + sql?: string +} + +type Props = { + engine: 'postgres' | 'trino' +} + +const ENGINE_META = { + postgres: { + title: 'PostgreSQL SQL Console', + sub: 'DB Vault · 10.0.21.51:5432 · user mo', + accent: 'text-blue-400', + }, + trino: { + title: 'Trino SQL Console', + sub: 'Lakehouse · 10.0.21.50:8089 · federated lakehouse queries', + accent: 'text-violet-400', + }, +} + +export function SqlWorkbench({ engine }: 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) + const [error, setError] = useState(null) + + const loadSamples = useCallback(async () => { + const r = await fetch(`/api/sql/samples/${engine}`) + if (r.ok) { + const j = await r.json() + setSamples(j.samples || []) + if (j.samples?.[0]) setSql(j.samples[0].sql) + } + }, [engine]) + + useEffect(() => { + loadSamples() + setResult(null) + setBenchmark(null) + setError(null) + }, [engine, loadSamples]) + + const run = async (query?: string) => { + const q = (query ?? sql).trim() + if (!q) return + setLoading(true) + setError(null) + try { + const r = await fetch('/api/sql/execute', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ engine, sql: q }), + }) + const j = await r.json() + if (!r.ok || !j.ok) { + setError(j.error || 'Query failed') + setResult(null) + return + } + setSql(q) + setResult(j) + } catch { + setError('SQL API unavailable') + } finally { + setLoading(false) + } + } + + const runBenchmark = async () => { + setBenchLoading(true) + setError(null) + try { + const r = await fetch('/api/sql/benchmark', { method: 'POST' }) + const j = await r.json() + if (!r.ok) { + setError(j.error || 'Benchmark failed') + return + } + setBenchmark(j) + } catch { + setError('Benchmark failed') + } finally { + setBenchLoading(false) + } + } + + return ( +
+
+
+

+ + {meta.title} +

+

{meta.sub}

+
+
+ + +
+
+ +
+ + +
+