"""
Operon LangGraph Visualizer -- Per-Stage Graph Topology
========================================================
Build a multi-stage organism, compile it to a per-stage LangGraph,
visualize the graph topology, and run it to see which stages execute.
Run locally: pip install gradio && python space-langgraph-visualizer/app.py
"""
import html as html_mod
import sys
from pathlib import Path
import gradio as gr
try:
_repo_root = Path(__file__).resolve().parents[2]
if str(_repo_root) not in sys.path:
sys.path.insert(0, str(_repo_root))
except IndexError:
pass # Running on HF — operon-ai installed via requirements.txt
from operon_ai import ATP_Store, MockProvider, Nucleus, SkillStage, skill_organism
from operon_ai.convergence.langgraph_compiler import (
organism_to_langgraph,
run_organism_langgraph,
)
# ---------------------------------------------------------------------------
# Presets
# ---------------------------------------------------------------------------
PRESETS = {
"3-stage pipeline": {
"stages": [
("intake", "Normalizer", "fixed"),
("router", "Classifier", "fixed"),
("executor", "Engineer", "fuzzy"),
],
"task": "Fix the login crash after session timeout",
},
"4-stage incident": {
"stages": [
("triage", "Triager", "fixed"),
("classify", "Classifier", "fixed"),
("investigate", "Investigator", "fixed"),
("fix", "Engineer", "fuzzy"),
],
"task": "Production auth failures after JWT migration",
},
"2-stage simple": {
"stages": [
("analyze", "Analyst", "fixed"),
("respond", "Responder", "fuzzy"),
],
"task": "Summarize the quarterly report",
},
"5-stage deep": {
"stages": [
("intake", "Normalizer", "fixed"),
("triage", "Triager", "fixed"),
("plan", "Planner", "fuzzy"),
("execute", "Engineer", "deep"),
("review", "Reviewer", "fixed"),
],
"task": "Refactor the authentication middleware for SOC2 compliance",
},
}
# ---------------------------------------------------------------------------
# Visualization
# ---------------------------------------------------------------------------
def _node_html(name, mode, index, total, executed=False, is_start=False, is_end=False):
"""Render a single graph node."""
if is_start:
return (
'')
if is_end:
return (
'')
mode_colors = {"fixed": "#3b82f6", "fuzzy": "#f59e0b", "deep": "#8b5cf6"}
border_color = mode_colors.get(mode, "#6b7280")
bg = f"{border_color}15"
check = ' ✓' if executed else ""
return (
f''
f'
'
f'
{html_mod.escape(name)}{check}
'
f'
mode: {html_mod.escape(mode)}
'
f'
'
+ "".join(nodes) + '
')
# Halt edges legend
halt_legend = (
''
'Each stage has a conditional edge: '
'continue → next stage, '
'halt/blocked → END'
'
')
return main_flow + halt_legend
# ---------------------------------------------------------------------------
# Core logic
# ---------------------------------------------------------------------------
def _parse_stages(stages_text):
"""Parse stages from text format: name, role, mode (one per line)."""
stages = []
for line in stages_text.strip().split("\n"):
line = line.strip()
if not line:
continue
parts = [p.strip() for p in line.split(",")]
if len(parts) >= 3:
stages.append((parts[0], parts[1], parts[2]))
elif len(parts) == 2:
stages.append((parts[0], parts[1], "fixed"))
elif len(parts) == 1:
stages.append((parts[0], parts[0].title(), "fixed"))
return stages
def visualize_and_run(stages_text, task, do_run):
if not stages_text.strip():
return "Enter at least one stage.
", ""
stages_info = _parse_stages(stages_text)
if not stages_info:
return "Could not parse stages. Use format: name, role, mode
", ""
# Build organism with deterministic handlers (avoids MockProvider
# substring-matching collisions across stages)
def _make_handler(stage_name, stage_role):
def handler(task, state, outputs, stage):
return f"[{stage_role}] Processed: task complete."
return handler
fast = Nucleus(provider=MockProvider(responses={}))
deep = Nucleus(provider=MockProvider(responses={}))
org = skill_organism(
stages=[
SkillStage(name=name, role=role,
handler=_make_handler(name, role),
mode=mode)
for name, role, mode in stages_info
],
fast_nucleus=fast,
deep_nucleus=deep,
budget=ATP_Store(budget=2000, silent=True),
)
# Compile to LangGraph
graph = organism_to_langgraph(org)
all_nodes = list(graph.nodes.keys())
stage_nodes = [n for n in all_nodes if not n.startswith("__")]
# Graph stats
stats_html = (
f''
f'Graph Stats: '
f'{len(stage_nodes)} stage nodes, '
f'{len(stage_nodes)} conditional edges (continue/halt), '
f'1 START edge, 1 terminal edge'
f'
')
# Visualize
executed_stages = []
run_html = ""
if do_run and task.strip():
result = run_organism_langgraph(org, task=task.strip())
executed_stages = result.metadata.get("stages_completed", [])
# Build run results
rows = ""
for sr in result.stage_outputs.items():
name, output = sr
preview = str(output)[:60]
rows += (
f''
f'| {html_mod.escape(name)} | '
f'{html_mod.escape(preview)} |
')
cert_rows = ""
for cv in result.certificates_verified:
status = "HOLDS" if cv["holds"] else "FAILS"
color = "#22c55e" if cv["holds"] else "#ef4444"
cert_rows += (
f''
f'{cv["theorem"]}: {status}')
run_html = (
f''
f'
'
f'Execution Results '
f''
f'({result.timing_ms:.1f} ms)
'
f'
'
f'
'
f''
f'| Stage | '
f'Output |
'
f'{rows}
'
f'
{cert_rows or "No certificates"}
'
f'
'
f'Halted: {result.metadata.get("halted", False)}
'
f'