""" 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)