π Judge Quickstart (90 seconds)
- Paste this or use your own:
Design a 3-site retail network with guest WiFi, POS VLANs, cameras, and secure VPN to HQ.
- Click Run Full Pipeline (keep simulation/read-only on).
- Review Status + tabs: Source of Truth YAML β BOM β Setup Guide.
- Peek at Live API Activity Monitor to see LLM/tool calls and costs.
Secrets: set BLAXEL_MCP_1ST_BDAY (or Anthropic/OpenAI/etc.) and optional GNS3 env vars for brownfield. Build info below shows deploy mode.
""",
)
gr.Markdown(dev_info())
device_choices = topology.list_devices()
# ===== API MONITORING DASHBOARD =====
# This is the key feature judges will love to see!
gr.Markdown("""
## π Live API Activity Monitor
**Real-time visibility into all API calls - LLM, GNS3, and more**
""")
with gr.Row(elem_classes=["og-main-row"]):
with gr.Column(scale=1, elem_classes=["og-panel"]):
gr.Markdown("### π Session Statistics")
api_stats = gr.Markdown(value=monitor.get_stats().format_dashboard(include_heading=False))
refresh_stats_btn = gr.Button("π Refresh Stats", size="sm", variant="secondary")
with gr.Column(scale=2, elem_classes=["og-panel"]):
gr.Markdown("### π‘ Recent API Calls")
api_activity = gr.Markdown(value=monitor.format_activity_feed())
refresh_activity_btn = gr.Button("π Refresh Activity", size="sm", variant="secondary")
with gr.Column(scale=2, elem_classes=["og-panel"]):
gr.Markdown("### π Runtime Log (tail)")
log_output = gr.Textbox(value="", lines=12, label="Latest Logs", interactive=False)
refresh_log_btn = gr.Button("π Refresh Logs", size="sm", variant="secondary")
# Auto-refresh handlers
def refresh_api_stats():
return monitor.get_stats().format_dashboard(include_heading=False)
def refresh_api_activity():
return monitor.format_activity_feed()
refresh_stats_btn.click(fn=refresh_api_stats, outputs=[api_stats])
refresh_activity_btn.click(fn=refresh_api_activity, outputs=[api_activity])
def refresh_logs():
try:
log_path = Path("app.log")
if not log_path.exists():
return "No logs yet."
import subprocess
tail = subprocess.run(["tail", "-n", "60", str(log_path)], capture_output=True, text=True)
return tail.stdout or "No logs yet."
except Exception as e:
return f"Failed to read logs: {e}"
refresh_log_btn.click(fn=refresh_logs, outputs=[log_output])
# Auto-refresh runtime log every 5s
log_timer = gr.Timer(5.0)
log_timer.tick(fn=refresh_logs, outputs=[log_output])
gr.Markdown("---")
# ===== INTERACTIVE CONSULTATION TAB =====
# Temporarily disabled due to Gradio 4.44.0 schema bug
# TODO: Re-enable when upgrading to Gradio 4.44.1+
# gr.Markdown("""### π¬ Interactive Consultation
# **Have a conversation before running the full pipeline**
#
# Not sure what you need? Start here! The AI will ask clarifying questions to understand your network requirements.
# """)
gr.Markdown("---")
# Main Pipeline UI
gr.Markdown("""### πΏ Network Automation Pipeline
**From Consultation β Production**
Describe your network in plain English. Overgrowth handles the rest:
1. π¬ **Consultation** - Capture requirements
2. π **Source of Truth** - Generate network data model
3. π **Diagram** - Visualize topology
4. π **Bill of Materials** - Hardware shopping list
5. π§ **Setup Guide** - Human deployment steps + OOB network
6. π€ **Autonomous Deploy** - AI configures devices (Ansible/Netmiko)
7. ποΈ **Observability** - Monitoring, SNMP, topology discovery
8. β
**Validation** - pytest-based network testing & compliance
_Brownfield ready: import existing NetBox/GNS3/SoT data, run simulation-only, and keep validation read-only._
""")
with gr.Row(elem_classes=["og-main-row"]):
with gr.Column(scale=1, elem_classes=["og-panel"]):
pipeline_input = gr.Textbox(
label="Describe Your Network",
placeholder=(
"Example: We're a coffee shop chain with 3 locations. "
"We need WiFi for customers, POS systems with payment processing, "
"security cameras, and secure VPN to HQ for centralized management. "
"Each location has ~50 customers at peak time."
),
lines=10,
)
mode_choice = gr.Dropdown(
label="Pipeline Mode",
choices=[
"Design new (greenfield)",
"Import from NetBox",
"Import from GNS3",
"Load existing SoT file"
],
value="Design new (greenfield)"
)
gr.Markdown("**Optional: provide an API key to override defaults for this run**")
anthropic_key_box = gr.Textbox(
label="Anthropic API Key (optional)",
placeholder="sk-ant-...",
type="password",
)
openai_key_box = gr.Textbox(
label="OpenAI API Key (optional)",
placeholder="sk-...",
type="password",
)
gns3_project_input = gr.Dropdown(
label="GNS3 project (brownfield)",
choices=project_choices or ["overgrowth"],
value=project_choices[0] if project_choices else "overgrowth",
visible=gns3_enabled,
interactive=gns3_enabled,
)
if not gns3_enabled:
gr.Markdown(
"> GNS3 not configured in this Space; pipeline will run simulation-only and hide brownfield controls.",
elem_classes=["og-label"]
)
sot_path_input = gr.Textbox(
label="Existing SoT path",
value="infra/network_model.yaml"
)
simulation_only = gr.Checkbox(
label="Simulation-only (skip deployment)",
value=True
)
read_only_validation = gr.Checkbox(
label="Read-only validation (no remediation)",
value=True
)
run_pipeline_btn = gr.Button("π Run Full Pipeline", variant="primary", size="lg")
with gr.Column(scale=2, elem_classes=["og-panel", "og-tabs"]):
gr.Markdown("#### π¦ Pipeline Outputs")
pipeline_status = gr.Markdown(label="Status")
sot_output = gr.Code(label="Network Model (YAML)", language="yaml", lines=20)
bom_output = gr.Markdown(label="Shopping List")
setup_output = gr.Markdown(label="Deployment Guide")
# Pipeline handler
def run_pipeline(user_input, mode_choice, gns3_project, sot_path, simulation_only_flag, read_only_flag, anthropic_key, openai_key):
"""Execute the full automation pipeline"""
from agent.pipeline_engine import OvergrowthPipeline
from agent.llm_client import LLMClient
# Apply per-run API keys (not persisted)
if anthropic_key:
os.environ["ANTHROPIC_API_KEY"] = anthropic_key.strip()
os.environ["ANTHROPIC_MCP_1ST_BDAY"] = anthropic_key.strip()
os.environ["OG_LLM_PROVIDER"] = "anthropic"
elif openai_key:
os.environ["OPENAI_API_KEY"] = openai_key.strip()
os.environ["OPENAI_MCP_1ST_BDAY"] = openai_key.strip()
os.environ["OG_LLM_PROVIDER"] = "openai"
pipeline = OvergrowthPipeline()
try:
# Report LLM env status up-front so users see if keys are missing
llm_env = LLMClient.env_status()
# If user explicitly wants deployment (simulation_only_flag=False), ensure we drop read-only guard.
if simulation_only_flag is False:
read_only_flag = False
# Run the pipeline (this will generate API calls that get tracked)
mode_map = {
"Design new (greenfield)": None,
"Import from NetBox": "netbox",
"Import from GNS3": "gns3",
"Load existing SoT file": "file",
}
chosen_source = mode_map.get(mode_choice)
source_note = ""
if not gns3_enabled and chosen_source == "gns3":
chosen_source = None
source_note = "\n> GNS3 not configured; running greenfield simulation instead.\n"
simulation_only_flag = True
if chosen_source:
results = pipeline.run_brownfield_pipeline(
consultation_input=user_input,
source=chosen_source,
project_name=gns3_project,
sot_path=sot_path,
simulation_only=simulation_only_flag,
read_only_validation=read_only_flag,
)
else:
results = pipeline.run_full_pipeline(user_input)
# If the pipeline stopped early for clarifications, show questions and exit
if results.get("needs_more_input"):
status = "## π€ Consultation Needed\n\n"
status += "Your request is too brief. Please answer these questions to continue:\n\n"
for q in results.get('questions', []):
status += f"- {q}\n"
return status, "", "", "", "", monitor.get_stats().format_dashboard(include_heading=False), monitor.format_activity_feed()
# Check pre-flight validation
preflight = results.get('preflight', {})
ready_to_deploy = preflight.get('ready_to_deploy', False)
# Format status with API usage summary
status = "## π Pipeline Execution Complete!\n\n"
if source_note:
status += source_note + "\n"
status += "> Diagrams disabled here; open the GNS3 project link for topology view.\n\n"
# Add API usage summary at the top
stats = monitor.get_stats()
status += f"### π° API Usage This Session\n"
status += f"- **Total Cost:** ${stats.total_cost:.4f}\n"
status += f"- **LLM Calls:** {stats.llm_calls} | **GNS3 Calls:** {stats.gns3_calls}\n"
status += f"- **Tokens:** {stats.total_tokens:,} ({stats.total_input_tokens:,} in / {stats.total_output_tokens:,} out)\n\n"
# LLM connectivity
status += "### π€ LLM Connectivity\n"
status += f"- Provider selected: **{llm_env.get('provider', 'unknown')}**\n"
status += f"- Anthropic key: {llm_env.get('anthropic_key')}\n"
status += f"- OpenAI key: {llm_env.get('openai_key')}\n"
status += f"- Blaxel key: {llm_env.get('blaxel_key', 'missing')}\n"
status += f"- SambaNova key: {llm_env.get('sambanova_key', 'missing')}\n"
status += f"- Nebius key: {llm_env.get('nebius_key', 'missing')}\n"
status += f"- HuggingFace key: {llm_env.get('huggingface_key', 'missing')}\n"
status += f"- Modal key: {llm_env.get('modal_key', 'missing')}\n\n"
wg_status = results.get("wg_status")
if wg_status:
status += "### πΈοΈ WireGuard\n"
status += f"- Status: {wg_status}\n\n"
commit = get_commit_hash()
status += "### π§Ύ Build\n"
status += f"- Commit: `{commit}`\n\n"
mode_info = results.get("mode")
if mode_info:
status += "### π³ Pipeline Mode\n"
status += f"- Brownfield source: **{mode_info.get('source', 'greenfield')}**\n"
status += f"- Simulation-only: {'Yes' if mode_info.get('simulation_only') else 'No'}\n"
status += f"- Read-only validation: {'Yes' if mode_info.get('read_only_validation') else 'No'}\n"
import_note = results.get("import_summary", {}).get("note") if isinstance(results.get("import_summary"), dict) else None
if import_note:
status += f"- Import notes: {import_note}\n"
status += "\n"
if stats.errors:
from agent.api_monitor import monitor as api_monitor_instance
recent_errors = api_monitor_instance.get_recent_errors(limit=3)
if recent_errors:
status += "### β οΈ Recent API Errors\n"
for err_call in reversed(recent_errors):
msg = err_call.error_message or "Unknown error"
status += f"- {err_call.provider.upper()} {err_call.endpoint}: {msg}\n"
status += "\n"
# Budget visibility
if stats.budget_limit not in (None, 0):
remaining = stats.budget_remaining()
status += f"- **Budget:** ${stats.budget_limit:.2f} (remaining ${remaining:.2f})\n"
budget_note = stats.budget_status()
if stats.budget_alert_triggered or (stats.budget_usage_ratio() or 0) >= stats.budget_alert_fraction:
status += f"> π¨ {budget_note}\n\n"
else:
status += f" - {budget_note}\n\n"
# Lab simulation link (GNS3)
lab_info = results.get("lab") or {}
gns3_base = lab_info.get("api") or os.getenv("GNS3_SERVER")
gns3_project = lab_info.get("project_name") or os.getenv("GNS3_PROJECT_NAME", "overgrowth")
gns3_project_id = lab_info.get("project_id") or os.getenv("GNS3_PROJECT_ID")
gns3_web = lab_info.get("web_url")
if not gns3_web:
gns3_web_override = os.getenv("GNS3_WEB_URL")
if gns3_web_override:
gns3_web = gns3_web_override
elif gns3_base and gns3_project_id:
gns3_web = f"{gns3_base.rstrip('/')}/static/web-ui/server/1/project/{gns3_project_id}"
elif gns3_base:
gns3_web = f"{gns3_base.rstrip('/')}/static/webUi"
if gns3_base:
status += "### π§ͺ Lab Simulation\n"
status += f"- GNS3 API: {gns3_base}\n"
status += f"- Project: `{gns3_project}`\n"
if gns3_project_id:
status += f"- Project ID: `{gns3_project_id}`\n"
if gns3_web:
status += f"- Web UI: {gns3_web}\n"
missing_sites = []
if isinstance(lab_info, dict):
missing_sites = lab_info.get("missing_sites") or []
requested_sites = lab_info.get("requested_site_count") or results.get("requested_site_count")
builder_sites = lab_info.get("builder_site_count") or results.get("builder_site_count")
topology_sites = lab_info.get("topology_site_count") or results.get("topology_site_count")
actual_sites = lab_info.get("actual_site_count") or results.get("actual_site_count")
mismatch = lab_info.get("site_count_mismatch") or results.get("site_count_mismatch")
if requested_sites is not None:
status += f"- Requested site count (parsed): {requested_sites}\n"
if builder_sites is not None:
status += f"- Builder site count (GNS3): {builder_sites}\n"
if topology_sites is not None:
status += f"- Topology site count (telemetry): {topology_sites}\n"
if actual_sites is not None:
status += f"- Actual site count: {actual_sites}\n"
if mismatch:
status += (
f"- Site count mismatch: requested {mismatch.get('requested')}, "
f"builder {mismatch.get('builder')}, "
f"topology {mismatch.get('topology')}\n"
)
build_err = lab_info.get("build_error") or (lab_info.get("build_result") or {}).get("error")
if build_err:
status += f"- Build error: {build_err}\n"
if missing_sites:
status += f"- Missing sites in GNS3 build: {', '.join(missing_sites)}\n"
status += "\n"
# SSH preflight summary (if available)
preflight_ssh = results.get("deployment", {}).get("ssh_preflight") if isinstance(results.get("deployment"), dict) else None
if preflight_ssh:
reachable = len([r for r in preflight_ssh if r.get("status") == "reachable"])
unreachable = len([r for r in preflight_ssh if r.get("status") == "unreachable"])
status += "### π SSH Reachability\n"
status += f"- Reachable: {reachable}\n"
status += f"- Unreachable: {unreachable}\n\n"
# Pre-flight validation section
if ready_to_deploy:
status += "### β
Pre-flight Validation PASSED\n"
status += f"- Schema validation: β Passed\n"
status += f"- Policy checks: β Passed\n"
if preflight.get('warnings'):
status += f"- Warnings: {len(preflight['warnings'])}\n"
if preflight.get('info'):
status += f"- Info: {len(preflight['info'])}\n"
status += "\n"
else:
status += "### β Pre-flight Validation FAILED\n"
status += "**Deployment blocked until errors are fixed:**\n\n"
for error in preflight.get('errors', []):
status += f"- β {error}\n"
status += "\n"
if preflight.get('warnings'):
status += "**Warnings:**\n"
for warning in preflight.get('warnings', []):
status += f"- β οΈ {warning}\n"
status += "\n"
status += "### Validation Findings\n"
status += f"- Errors: {len(preflight.get('errors', []))}\n"
status += f"- Warnings: {len(preflight.get('warnings', []))}\n"
status += f"- Info: {len(preflight.get('info', []))}\n"
if preflight.get('batfish_analysis'):
bf = preflight['batfish_analysis']
status += f"- Batfish: {'mock' if bf.get('mock_mode') else 'analysis ran'} | Loops: {len(bf.get('routing_loops', []))} | Undefined refs: {len(bf.get('undefined_references', []))}\n"
status += "\n"
# Completed stages
status += "### Completed Stages:\n"
status += "0. " + ("β
" if ready_to_deploy else "β") + " Pre-flight Validation\n"
status += "1. β
Consultation - Intent captured\n"
status += "2. β
Source of Truth - Network model generated\n"
status += "3. π« Diagrams - Disabled (use GNS3 UI)\n"
status += "4. β
Bill of Materials - Shopping list ready\n"
status += "5. β
Setup Guide - Deployment instructions generated\n"
sim_only = results.get("mode", {}).get("simulation_only")
ro_mode = results.get("mode", {}).get("read_only_validation")
deploy_disabled = results.get("deployment_reason") == "deployment_disabled"
if ready_to_deploy:
if deploy_disabled:
status += "6. π« Autonomous Deploy - Disabled in this environment\n"
status += "7. ποΈ Observability - Snapshot only\n"
status += "8. β
Validation - Read-only checks complete\n\n"
elif sim_only or ro_mode:
status += "6. π§ͺ Autonomous Deploy - Skipped (simulation/read-only)\n"
status += "7. ποΈ Observability - Snapshot only\n"
status += "8. β
Validation - Read-only checks complete\n\n"
else:
status += "6. β³ Autonomous Deploy - Ready for execution\n"
status += "7. β³ Observability - Ready for setup\n"
status += "8. β³ Validation - Ready for verification\n\n"
else:
status += "6. π« Autonomous Deploy - BLOCKED (fix validation errors)\n"
status += "7. π« Observability - BLOCKED\n"
status += "8. π« Validation - BLOCKED\n\n"
status += "### π Files Created:\n"
status += "- `infra/network_model.yaml` - Source of Truth\n"
status += "- `infra/bill_of_materials.json` - BOM data\n"
status += "- `infra/setup_guide.md` - Deployment guide\n"
questions = results.get('questions', [])
if questions:
status += "\n### β Clarifying Questions\n"
for q in questions:
status += f"- {q}\n"
# Extract outputs
sot_yaml = results.get('model', {})
import yaml
sot_str = yaml.dump(sot_yaml, default_flow_style=False)
bom_md = results.get('shopping_list', 'No BOM generated')
setup_md = results.get('setup_guide', 'No setup guide generated')
updated_stats = monitor.get_stats().format_dashboard(include_heading=False)
updated_activity = monitor.format_activity_feed()
# Cache last known good outputs for offline fallback
try:
save_smoke_cache({
"status": status,
"sot": sot_str,
"bom": bom_md,
"setup": setup_md,
"stats": updated_stats,
"activity": updated_activity,
})
except Exception:
pass
return status, sot_str, bom_md, setup_md, updated_stats, updated_activity
except Exception as e:
cached = load_smoke_cache()
import traceback
error_block = f"\n\n```\n{traceback.format_exc()}\n```"
if cached:
status = f"## β οΈ Live pipeline failed\n\n{str(e)}\n\nShowing last known good snapshot instead."
status += error_block
return (
status,
cached.get("sot", ""),
cached.get("bom", ""),
cached.get("setup", ""),
cached.get("stats", monitor.get_stats().format_dashboard(include_heading=False)),
cached.get("activity", monitor.format_activity_feed()),
)
else:
error = f"## β Pipeline Error\n\n{str(e)}"
error += error_block
updated_stats = monitor.get_stats().format_dashboard(include_heading=False)
updated_activity = monitor.format_activity_feed()
return error, "", "", "", updated_stats, updated_activity
# Event Handlers
run_pipeline_btn.click(
fn=run_pipeline,
inputs=[
pipeline_input,
mode_choice,
gns3_project_input,
sot_path_input,
simulation_only,
read_only_validation,
anthropic_key_box,
openai_key_box,
],
outputs=[pipeline_status, sot_output, bom_output, setup_output, api_stats, api_activity],
)
return demo
demo = build_ui().queue() # enable queue by default for Spaces
app = demo # Hugging Face picks this up automatically
if __name__ == "__main__":
demo.launch(server_name="0.0.0.0", server_port=7860, share=False, show_api=False)