"""
Aurora Dashboard — A light, modern UI for BroJS agent task orchestration.
"""
import asyncio
import json
import os
import queue
import threading
import time
from datetime import datetime
os.environ["NO_PROXY"] = (
"openrouter.ai,platform.brojs.ru,git.brojs.ru,"
+ os.environ.get("NO_PROXY", "")
)
import streamlit as st
from dotenv import load_dotenv
from langchain_core.callbacks.base import BaseCallbackHandler
from langchain_core.messages import AIMessage, HumanMessage
load_dotenv()
st.set_page_config(
page_title="Aurora Dashboard",
page_icon="✨",
layout="wide",
initial_sidebar_state="collapsed",
)
GITEA_OWNER = os.getenv("GITEA_OWNER", "dapa46")
# ──── THEME & STYLING ────
st.markdown("""
""", unsafe_allow_html=True)
# ──── PAGE HEADER ────
st.markdown("""
✨ Aurora Dashboard
Streamlined task orchestration for BroJS. Send tasks, monitor progress, and control your course automation—all in one place.
""", unsafe_allow_html=True)
# ──── SESSION STATE ────
for key in ["chat_history", "chat_events", "chat_thread_id", "task_statuses"]:
if key not in st.session_state:
if key == "chat_thread_id":
st.session_state[key] = f"aurora-{int(time.time())}"
else:
st.session_state[key] = [] if key != "task_statuses" else {}
# ──── AGENT HELPERS ────
@st.cache_resource(show_spinner="🚀 Starting Aurora engine...")
def get_agent():
from src.agent.agent import agent
return agent
@st.cache_resource(show_spinner="📡 Loading orchestrator...")
def get_main_agent():
from src.agent.agent import agent
return agent
def _load_statuses():
from src.agent.mcp_client import load_journal_toolsets
async def _fetch():
j = load_journal_toolsets()
tools = {t.name: t for t in j.tasks_submissions_tools}
key = "mcp__journal-bh-professor__tasks_list"
tool = tools.get(key) or next((v for k, v in tools.items() if "tasks_list" in k), None)
if not tool:
return [], f"Task list tool not found."
raw = await tool.ainvoke({"courseId": "698b49da77cb6d4d2e43ce78"})
text = next((x["text"] for x in raw if x.get("type") == "text"), str(raw)) if isinstance(raw, list) else str(raw)
data = json.loads(text)
return (data.get("tasks", data) if isinstance(data, dict) else data), None
result = {"items": [], "error": None}
def _run():
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
result["items"], result["error"] = loop.run_until_complete(_fetch())
except Exception as e:
result["error"] = str(e)
finally:
loop.close()
t = threading.Thread(target=_run, daemon=True)
t.start()
t.join()
return result["items"], result["error"]
# ──── CALLBACKS & RUNNERS ────
_SOLVE_TOOLS = {"validate_teacher_comment", "generate_code_solution"}
_JOURNAL_PREFIX = "mcp__journal-bh-professor__"
class AgentCallback(BaseCallbackHandler):
def __init__(self, q: queue.Queue):
self.q = q
def _ts(self) -> str:
return datetime.now().strftime("%H:%M:%S")
def on_tool_start(self, serialized, input_str, **kwargs):
name = serialized.get("name", "?")
try:
args = json.loads(str(input_str)) if isinstance(input_str, str) else input_str
except:
args = {}
self.q.put({"t": "tool_start", "ts": self._ts(), "name": name, "args": args})
def on_tool_end(self, output, **kwargs):
self.q.put({"t": "tool_end", "ts": self._ts(), "output": str(output)[:280]})
def on_tool_error(self, error, **kwargs):
self.q.put({"t": "tool_error", "ts": self._ts(), "msg": str(error)[:200]})
def on_llm_start(self, *a, **kw):
self.q.put({"t": "thinking", "ts": self._ts()})
def on_llm_end(self, response, **kwargs):
try:
text = response.generations[0][0].text[:120]
self.q.put({"t": "llm_end", "ts": self._ts(), "preview": text})
except:
pass
def _run_agent_thread(agent, messages, config, q: queue.Queue, cb: AgentCallback):
async def _inner():
try:
result = await agent.ainvoke(messages, {**config, "callbacks": [cb]})
q.put({"t": "done", "result": result})
except Exception as e:
q.put({"t": "fatal", "msg": str(e)})
asyncio.run(_inner())
def _render_event(ev: dict) -> str:
ts = ev.get("ts", "")
kind = ev.get("t", "")
if kind == "thinking":
return f'💭 {ts} Thinking...
'
if kind == "tool_start":
name = ev["name"]
args = ev.get("args", {})
short = name.replace(_JOURNAL_PREFIX, "mcp::")
icon = "🔧"
if name in _SOLVE_TOOLS:
icon = "🧠"
short = f"[Subagent] {short}"
elif "gitea" in name:
icon = "📦"
elif "journal" in name:
icon = "📡"
hint = ""
for k in ("taskId", "path", "repo", "repo_name", "name"):
if k in args:
hint = f' — {args[k]}'
break
return f'{icon} {short}{hint}
'
if kind == "tool_end":
out = ev["output"].replace("<", "<").replace(">", ">")[:200]
return f'✓ {out}
'
if kind == "tool_error":
msg = ev["msg"].replace("<", "<")
return f'⚠ {msg}
'
if kind == "llm_end":
preview = ev.get("preview", "").replace("<", "<")[:100]
return f'✏ {ts} {preview}...
'
return ""
# ──── TABS ────
tab1, tab2, tab3 = st.tabs(["💬 Chat", "⚡ Pipeline", "📊 Status"])
# ═══════════════════════════════════════════════════════════════
# TAB 1: CHAT
# ═══════════════════════════════════════════════════════════════
with tab1:
st.subheader("Ask Aurora")
st.write("Chat with the agent. Ask questions, request task solutions, or check your course progress.")
# Display chat history
chat_col = st.container()
with chat_col:
for msg in st.session_state.chat_history:
if msg["role"] == "user":
st.markdown(f'{msg["text"]}
', unsafe_allow_html=True)
else:
st.markdown(f'{msg["text"]}
', unsafe_allow_html=True)
# Event log expander
if st.session_state.chat_events:
with st.expander(f"📋 Execution log ({len(st.session_state.chat_events)} events)", expanded=False):
html = "".join(_render_event(e) for e in st.session_state.chat_events[-60:])
st.markdown(f'{html}
', unsafe_allow_html=True)
# Input area
st.write("")
col1, col2 = st.columns([5, 1])
with col1:
user_msg = st.text_input("Message", placeholder="Ask about tasks, solve a problem, check status...", label_visibility="collapsed", key="chat_msg")
with col2:
send = st.button("Send", type="primary", use_container_width=True)
if send and user_msg.strip():
msg_text = user_msg.strip()
st.session_state.chat_history.append({"role": "user", "text": msg_text})
st.session_state.chat_events = []
messages = []
for m in st.session_state.chat_history:
if m["role"] == "user":
messages.append(HumanMessage(content=m["text"]))
else:
messages.append(AIMessage(content=m["text"]))
agent = get_agent()
event_q: queue.Queue = queue.Queue()
cb = AgentCallback(event_q)
all_events = []
ph_events = st.empty()
thread = threading.Thread(
target=_run_agent_thread,
args=(agent, {"messages": messages}, {"configurable": {"thread_id": st.session_state.chat_thread_id}}, event_q, cb),
daemon=True,
)
thread.start()
final = None
error = None
while thread.is_alive() or not event_q.empty():
while not event_q.empty():
ev = event_q.get_nowait()
if ev["t"] == "done":
final = ev["result"]
elif ev["t"] == "fatal":
error = ev["msg"]
else:
all_events.append(ev)
if all_events:
html = "".join(_render_event(e) for e in all_events[-50:])
ph_events.markdown(f'{html}
', unsafe_allow_html=True)
time.sleep(0.14)
ph_events.empty()
st.session_state.chat_events = all_events
if error:
st.session_state.chat_history.append({"role": "agent", "text": f"⚠️ Error: {error}"})
elif final:
msgs = final.get("messages", [])
reply = msgs[-1].content if msgs and hasattr(msgs[-1], "content") else "Complete."
st.session_state.chat_history.append({"role": "agent", "text": reply})
st.rerun()
if st.session_state.chat_history:
st.write("")
if st.button("🗑 Clear history", use_container_width=True):
st.session_state.chat_history = []
st.session_state.chat_events = []
st.session_state.chat_thread_id = f"aurora-{int(time.time())}"
st.rerun()
# ═══════════════════════════════════════════════════════════════
# TAB 2: PIPELINE
# ═══════════════════════════════════════════════════════════════
with tab2:
st.subheader("Automation Pipeline")
st.write("Execute a single task or run the full automation for all TODO items in KFU-26-1.")
col1, col2 = st.columns([3, 1])
with col1:
tid = st.text_input("Task ID", placeholder="Optional: enter a task ID to run just that one", label_visibility="visible", key="pipe_tid")
with col2:
st.write("")
run_single = st.button("▶ Run", type="primary", use_container_width=True)
if run_single:
if tid.strip():
task_id = tid.strip()
prompt = f"Solve task taskId={task_id} course 698b49da77cb6d4d2e43ce78"
repo_url = f"https://git.brojs.ru/{GITEA_OWNER}/task-{task_id}"
agent = get_agent()
event_q: queue.Queue = queue.Queue()
cb = AgentCallback(event_q)
st.info(f"Running task {task_id}...")
ph_log = st.empty()
ph_result = st.empty()
all_events = []
thread = threading.Thread(
target=_run_agent_thread,
args=(agent, {"messages": [HumanMessage(content=prompt)]}, {"configurable": {"thread_id": f"pipe-{task_id}"}}, event_q, cb),
daemon=True,
)
thread.start()
while thread.is_alive() or not event_q.empty():
while not event_q.empty():
ev = event_q.get_nowait()
if ev["t"] not in ("done", "fatal"):
all_events.append(ev)
if all_events:
html = "".join(_render_event(e) for e in all_events[-50:])
ph_log.markdown(f'{html}
', unsafe_allow_html=True)
time.sleep(0.15)
ph_log.empty()
ph_result.success(f"✅ Task complete! [Open in Gitea]({repo_url})")
else:
st.warning("Enter a Task ID first.")
st.write("")
st.divider()
st.subheader("Full Automation")
st.write("Run the agent orchestrator to execute all TODO tasks in sequence.")
if st.button("⚡ Run Full Pipeline", use_container_width=True):
main_ag = get_main_agent()
event_q: queue.Queue = queue.Queue()
cb = AgentCallback(event_q)
state = {"result": None, "error": None}
def _run_full():
async def _inner():
try:
state["result"] = await main_ag.ainvoke(
{"messages": [HumanMessage(content=(
"Execute all tasks with status todo in course KFU-26-1 "
"(courseId=698b49da77cb6d4d2e43ce78). "
"Get the list, solve each one sequentially, report results."
))]},
{
"configurable": {"thread_id": f"aurora-all-{int(time.time())}"},
"callbacks": [cb],
},
)
except Exception as e:
state["error"] = str(e)
asyncio.run(_inner())
ph_spin = st.empty()
ph_log = st.empty()
all_events = []
thread = threading.Thread(target=_run_full, daemon=True)
thread.start()
with ph_spin.container():
with st.spinner("Orchestrator running..."):
while thread.is_alive() or not event_q.empty():
while not event_q.empty():
ev = event_q.get_nowait()
if ev["t"] not in ("done", "fatal"):
all_events.append(ev)
if all_events:
html = "".join(_render_event(e) for e in all_events[-60:])
ph_log.markdown(f'{html}
', unsafe_allow_html=True)
time.sleep(0.14)
ph_spin.empty()
if state["error"]:
st.error(f"Error: {state['error']}")
else:
msgs = state["result"].get("messages", []) if state["result"] else []
reply = msgs[-1].content if msgs and hasattr(msgs[-1], "content") else "Complete."
st.success(f"✅ Automation done: {reply[:400]}")
# ═══════════════════════════════════════════════════════════════
# TAB 3: STATUS
# ═══════════════════════════════════════════════════════════════
with tab3:
st.subheader("Course Status")
st.write("View task statuses for KFU-26-1.")
if st.button("🔄 Refresh", type="primary", use_container_width=True):
items, err = _load_statuses()
if err:
st.error(err)
else:
st.session_state.task_statuses = items
search = st.text_input("Search tasks", placeholder="Filter by ID, title, or status...", label_visibility="collapsed")
items = st.session_state.task_statuses
if not items:
st.info("Click Refresh to load task data.")
else:
rows = []
for item in items:
task = item.get("task", item) if isinstance(item, dict) else {}
tid = task.get("id", "")
status = item.get("status", "")
title = task.get("title", task.get("name", ""))
if search and search.lower() not in (tid + title + status).lower():
continue
rows.append({
"ID": tid[:12] + "...",
"Title": title,
"Status": status,
"Repo": f"[Link](https://git.brojs.ru/{GITEA_OWNER}/task-{tid})",
})
st.dataframe(rows, use_container_width=True, hide_index=True)
st.divider()
counts = {}
for row in rows:
s = row["Status"]
counts[s] = counts.get(s, 0) + 1
cols = st.columns(len(counts))
for col, (status, count) in zip(cols, counts.items()):
col.metric(status.replace("_", " ").title(), count)
st.markdown('', unsafe_allow_html=True)