feat(dashboard): página Ejecutar con trace timeline y violations rendering

- components/trace_view.render_trace: decision_path como timeline con duración.
- components/violation_view.render_violations: violaciones con badge de severidad.
- pages/2_▶️_Ejecutar.py: selector de agente, botones de escenarios pregrabados,
  textarea, invoke y render de status/output/violations/traza. Aviso si queda en
  awaiting_approval. (Se omite la variable muerta chosen_example del plan.)

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
Juan
2026-05-11 12:42:12 +02:00
co-authored by Claude Opus 4.7
parent f2ef01d98d
commit ab21db2476
3 changed files with 122 additions and 0 deletions
@@ -0,0 +1,24 @@
"""Renderiza el decision_path como timeline."""
from __future__ import annotations
import streamlit as st
def render_trace(decision_path: list[dict]) -> None:
if not decision_path:
st.info("Aún no hay traza disponible.")
return
st.subheader("📜 Decision path")
for step in decision_path:
with st.container(border=True):
cols = st.columns([2, 1, 1])
with cols[0]:
st.markdown(f"**{step['step']}**")
with cols[1]:
st.caption(step.get("timestamp", ""))
with cols[2]:
st.metric("ms", step.get("duration_ms", 0), label_visibility="collapsed")
if step.get("detail"):
with st.expander("detalle"):
st.json(step["detail"])
@@ -0,0 +1,26 @@
"""Renderiza violaciones de guardrails con badges de severidad."""
from __future__ import annotations
import streamlit as st
_SEV_BADGE = {
"info": "🟦",
"warning": "🟧",
"block": "🟥",
}
def render_violations(violations: list[dict]) -> None:
if not violations:
st.success("Sin violaciones.")
return
st.subheader("🛡️ Violaciones de guardrails")
for v in violations:
with st.container(border=True):
badge = _SEV_BADGE.get(v["severity"], "")
st.markdown(
f"{badge} **{v['validator']}** — `{v['stage']}` — "
f"severity=`{v['severity']}` — blocked=`{v['blocked']}`"
)
st.caption(v["message"])
@@ -0,0 +1,72 @@
"""Página para invocar un agente y visualizar la traza completa."""
from __future__ import annotations
from pathlib import Path
import streamlit as st
from agentforge_dashboard.client import CoreClient
from agentforge_dashboard.components.trace_view import render_trace
from agentforge_dashboard.components.violation_view import render_violations
@st.cache_resource
def get_client() -> CoreClient:
return CoreClient()
client = get_client()
st.title("▶️ Ejecutar Agente")
st.caption("Lanza una ejecución con la cadena completa de gobierno.")
agents = client.list_agents()
if not agents:
st.warning("No hay agentes registrados.")
st.stop()
names = [a["name"] for a in agents]
agent_name = st.selectbox("Agente", names)
# Cargar escenarios pregrabados si existen en el disco montado.
examples_dir = Path("agents") / agent_name / "examples"
example_files = sorted(examples_dir.glob("*.txt")) if examples_dir.exists() else []
col_l, col_r = st.columns([3, 1])
with col_r:
st.markdown("**Escenarios**")
for ef in example_files:
if st.button(ef.stem, use_container_width=True):
st.session_state["input_text"] = ef.read_text(encoding="utf-8")
with col_l:
user_input = st.text_area(
"Descripción del incidente",
height=240,
key="input_text",
placeholder="Pega aquí la descripción del incidente o usa un escenario...",
)
if st.button("🚀 Invocar agente", type="primary", disabled=not user_input):
with st.spinner("Ejecutando..."):
result = client.invoke_agent(agent_name, {"input": user_input})
st.session_state["last_execution"] = result
execution = st.session_state.get("last_execution")
if execution and "error" not in execution:
st.divider()
st.markdown(f"**trace_id:** `{execution['trace_id']}`")
st.markdown(f"**Status:** `{execution['status']}`")
if execution["status"] == "awaiting_approval":
st.warning(
"Esta ejecución requiere aprobación humana. "
"Ve a la página **Aprobaciones** para revisar y decidir."
)
if execution.get("final_output"):
st.subheader("📦 Output final")
st.json(execution["final_output"])
render_violations(execution.get("violations", []))
render_trace(execution.get("decision_path", []))
elif execution and "error" in execution:
st.error(f"Error de la API: {execution['error']}")