# ui.py — Visual agent dashboard for PipelineEnv
# Mounts a Gradio app at /ui on top of the FastAPI server.
# Uses a built-in deterministic agent — no external LLM required.
import json
import os
import time
import gradio as gr
from server.pipeline_environment import PipelineEnvironment
from models import PipelineAction, RepairAction
# ── Config ─────────────────────────────────────────────
USE_LLM = os.getenv("USE_LLM", "").lower() == "true"
API_BASE_URL = os.getenv("API_BASE_URL") or "https://router.huggingface.co/v1"
MODEL_NAME = os.getenv("MODEL_NAME") or "meta-llama/Llama-3.1-8B-Instruct"
API_KEY = os.getenv("HF_TOKEN") or os.getenv("API_KEY") or "dummy"
env = PipelineEnvironment()
# Deterministic action plans for each task
ACTION_PLANS = {
"easy": ["fix_test"],
"medium": ["fix_docker_config", "set_env_var"],
"hard": ["rollback_commit", "add_dependency", "fix_yaml_config"],
}
ACTION_ICONS = {
"fix_test": "\U0001f9ea",
"set_env_var": "\U0001f511",
"fix_docker_config": "\U0001f40b",
"fix_yaml_config": "\U0001f4c4",
"retry_stage": "\U0001f504",
"rollback_commit": "\u23ea",
"add_dependency": "\U0001f4e6",
"no_op": "\U0001f4a4",
}
# ── Helpers ────────────────────────────────────────────
def _render_env(obs, step, max_s):
lines = []
name = obs.get("pipeline_name", "pipeline")
lines.append(
f''
f'\U0001f4e6 Pipeline: {name} | Step {step}/{max_s}
'
)
for s in obs.get("stages", []):
icon = {"passing": "\u2705", "failing": "\u274c", "skipped": "\u23ed\ufe0f"}.get(s["status"], "?")
err = f' ({s["error"]})' if s.get("error") else ""
lines.append(f'{icon} {s["name"]:<10} {s["status"]:<10}{err}')
return "
".join(lines)
def _render_health(score):
pct = int(score * 100)
color = "#3fb950" if score >= 0.8 else "#d29922" if score >= 0.4 else "#f85149"
return (
f''
f'Health: {score:.2f} / 1.0 ({pct}%)
'
)
def _deterministic_action(task_id, step_num):
"""Built-in agent that knows the correct action sequence."""
plan = ACTION_PLANS.get(task_id, [])
if step_num - 1 < len(plan):
return {"action": plan[step_num - 1], "target": None, "value": None}
return {"action": "no_op", "target": None, "value": None}
# ── Yielding generator ──────────────────────────────────
def run_task(task_id):
obs = env.reset(task_id).model_dump()
max_s = obs["max_steps"]
term_lines = []
action_log = []
rewards = []
success = False
current_score = 0.0
# reset banner
term_lines.append("\U0001f680 RESET \u2014 starting episode")
term_lines.append(f" Task : {task_id}")
term_lines.append(f" Desc : {obs['task_description']}")
term_lines.append(f" Steps : {max_s}")
term_lines.append(f" Health: {obs['health_score']:.2f}")
def _snapshot():
te = "
".join(term_lines)
term_box = f'{te}
'
al = "
".join(f'{a}' for a in action_log) if action_log else 'Waiting\u2026'
sm = f'Running\u2026 step {len(action_log)}/{max_s} | health {obs["health_score"]:.2f} | score {current_score:.3f}'
return _render_env(obs, len(action_log), max_s), _render_health(obs["health_score"]), term_box, al, sm
yield _snapshot()
time.sleep(0.4)
for step_num in range(1, max_s + 1):
# Use deterministic agent
ad = _deterministic_action(task_id, step_num)
aname = ad.get("action", "no_op")
icon = ACTION_ICONS.get(aname, "\u2699")
action_log.append(f"{icon} Step {step_num}: {aname}")
term_lines.append(f' \u279c Action \u279c {aname}')
try:
ra = getattr(RepairAction, aname)
except Exception:
ra = RepairAction.no_op
result = env.step(PipelineAction(action=ra, target=ad.get("target"), value=ad.get("value")))
rwd = result.get("reward", 0) or 0.0
done = result.get("done", False)
obs = result.get("observation", obs)
info = result.get("info", {})
current_score = info.get("grader_score", current_score)
rewards.append(rwd)
if rwd > 0:
term_lines.append(f' reward {rwd:+.3f} health \u2192 {obs["health_score"]:.2f}')
elif rwd < 0:
term_lines.append(f' reward {rwd:+.3f} health \u2192 {obs["health_score"]:.2f}')
else:
term_lines.append(f' reward {rwd:+.3f} health \u2192 {obs["health_score"]:.2f}')
yield _snapshot()
time.sleep(0.3)
if done:
success = current_score >= 0.99
break
# --- final summary ---
total_r = sum(rewards)
avg_r = total_r / len(rewards) if rewards else 0
sc = "#3fb950" if success else "#f85149"
st = "\u2705 PIPELINE HEALED" if success else "\u274c FAILED TO HEAL"
term_lines.append(f'{st}')
term_lines.append(f' Steps: {len(action_log)} | Total reward: {total_r:+.3f} | Score: {current_score:.3f}')
term_lines.append(f' Agent: Deterministic (built-in)')
yield _snapshot()
# ── Gradio layout ──────────────────────────────────────
with gr.Blocks(title="PipelineEnv \u2014 Agent Dashboard") as demo:
gr.HTML('')
gr.HTML(
''
'
\U0001f527 PipelineEnv
'
'
RL Agent \u2022 CI/CD Self-Healing Dashboard
PIPELINE STAGES
')
pipeline_disp = gr.HTML('Select a task and click Run Agent.
')
gr.HTML('HEALTH
')
health_disp = gr.HTML("")
with gr.Column(scale=1):
gr.HTML('ACTION LOG
')
action_disp = gr.HTML('Waiting\u2026
')
gr.HTML('SUMMARY
')
summary_disp = gr.HTML("")
gr.HTML('TERMINAL
')
terminal_disp = gr.HTML("")
run_btn.click(
fn=run_task,
inputs=[task_dd],
outputs=[pipeline_disp, health_disp, terminal_disp, action_disp, summary_disp],
)
if __name__ == "__main__":
demo.launch(server_port=7861)