"""dashboard/reasoning.py — render the agent's full reasoning trace. The trace is a list of step dicts, built incrementally during graph streaming and rendered (live during generation, persistently afterwards) inside an expander on the Verdict tab. Step shapes: {"kind": "assistant_text", "text": str} {"kind": "tool_call", "id": str, "name": str, "args": dict, "status": "pending"|"done"} {"kind": "tool_result", "tool_call_id": str, "name": str, "snippet": str, "full": str} {"kind": "synthesis", "status": "in_progress"|"done", "error": str | None} """ from __future__ import annotations import json from typing import Any import streamlit as st from dashboard.theme import ( BG, BG_MUTED, BORDER, BORDER_STRONG, GREEN, RED, AMBER, BLUE, TEXT, TEXT_MUTED, INFO_BG, INFO_BORDER, WARN_BG, WARN_BORDER, ) from dashboard.components import source_badge SNIPPET_LEN = 200 # ────────────────────────────────────────────────────────────────────────────── # Trace builder helpers — called from app.py during graph.stream # ────────────────────────────────────────────────────────────────────────────── def _stringify(content: Any) -> str: if isinstance(content, str): return content if isinstance(content, list): parts = [] for block in content: if isinstance(block, dict) and block.get("type") == "text": parts.append(block.get("text", "")) elif isinstance(block, str): parts.append(block) return "\n".join(p for p in parts if p) return str(content) def absorb_agent_messages(trace: list[dict], messages: list) -> None: """Append assistant_text + tool_call steps from a batch of agent messages.""" for msg in messages: text = _stringify(getattr(msg, "content", "")).strip() if text: trace.append({"kind": "assistant_text", "text": text}) for tc in getattr(msg, "tool_calls", None) or []: trace.append({ "kind": "tool_call", "id": tc.get("id", ""), "name": tc.get("name", ""), "args": tc.get("args") or {}, "status": "pending", }) def absorb_tool_messages(trace: list[dict], messages: list) -> None: """Append tool_result steps and mark matching tool_call steps done.""" for msg in messages: tcid = getattr(msg, "tool_call_id", "") name = getattr(msg, "name", "") or "" full = _stringify(getattr(msg, "content", "")) snippet = full.strip().replace("\n", " ") if len(snippet) > SNIPPET_LEN: snippet = snippet[:SNIPPET_LEN] + "…" trace.append({ "kind": "tool_result", "tool_call_id": tcid, "name": name, "snippet": snippet, "full": full, }) for step in trace: if step["kind"] == "tool_call" and step["id"] == tcid: step["status"] = "done" def mark_synthesis(trace: list[dict], status: str, error: str | None = None) -> None: """Add or update a synthesis step.""" for step in trace: if step["kind"] == "synthesis": step["status"] = status step["error"] = error return trace.append({"kind": "synthesis", "status": status, "error": error}) # ────────────────────────────────────────────────────────────────────────────── # Progress estimation — called from app.py live fragment # ────────────────────────────────────────────────────────────────────────────── def estimate_progress(trace: list[dict]) -> tuple[float, str]: """Return (0.0–1.0, status label) based on current trace state.""" if not trace: return 0.02, "Starting…" for step in trace: if step["kind"] == "synthesis": if step["status"] == "done": return 1.0, "Brief ready — switching to Verdict tab" return 0.92, "Synthesizing brief…" completed = sum(1 for s in trace if s["kind"] == "tool_call" and s["status"] == "done") pending = sum(1 for s in trace if s["kind"] == "tool_call" and s["status"] == "pending") if completed == 0 and pending > 0: return 0.06, "Running first queries…" value = 0.10 + min(completed / 10, 1.0) * 0.78 return min(value, 0.88), f"Round {completed} — investigating…" # ────────────────────────────────────────────────────────────────────────────── # Pre-rendering: pair tool_call + tool_result into unified "tool_round" steps # ────────────────────────────────────────────────────────────────────────────── def _pair_steps(trace: list[dict]) -> list[dict]: """Walk the flat trace and merge each tool_call with its matching tool_result.""" results_by_id: dict[str, dict] = {} for step in trace: if step["kind"] == "tool_result": results_by_id[step["tool_call_id"]] = step paired: list[dict] = [] for step in trace: kind = step["kind"] if kind == "assistant_text": paired.append(step) elif kind == "tool_call": paired.append({ "kind": "tool_round", "call": step, "result": results_by_id.get(step["id"]), }) elif kind == "tool_result": pass # consumed above elif kind == "synthesis": paired.append(step) return paired # ────────────────────────────────────────────────────────────────────────────── # Helpers # ────────────────────────────────────────────────────────────────────────────── _TOOL_CATEGORIES: dict[str, str] = { "search_filing": "filing", "get_financial_metrics": "filing", "get_analyst": "filing", "search_transcript": "transcript", "get_news": "news", "search_news": "news", } def _tool_category(name: str) -> str: for key, cat in _TOOL_CATEGORIES.items(): if key in name: return cat return "tool" def _fmt_args(args: dict, max_len: int = 120) -> str: if not args: return "" try: s = json.dumps(args, ensure_ascii=False, separators=(", ", ": "), default=str).strip("{}") except Exception: s = str(args) return s if len(s) <= max_len else s[: max_len - 1] + "…" def _trunc(text: str, n: int = 75) -> str: text = text.strip().replace("\n", " ") return text if len(text) <= n else text[:n - 1] + "…" def _details_block(full: str) -> str: """HTML
collapsible for full tool result. Empty if content fits in snippet.""" if not full or len(full.strip()) <= SNIPPET_LEN: return "" escaped = full.replace("&", "&").replace("<", "<").replace(">", ">") return ( f'
' f'' f'▶ Show full result' f'' f'
{escaped}
' f'
' ) def _details_wrap(summary_html: str, body_html: str, is_open: bool) -> str: """Wrap body_html in a
block with the given summary.""" open_attr = " open" if is_open else "" return ( f'' f'' f'{summary_html}' f'' f'{body_html}' f'
' ) # ────────────────────────────────────────────────────────────────────────────── # Step renderers # ────────────────────────────────────────────────────────────────────────────── def _render_assistant_text(step: dict, is_recent: bool = True) -> None: preview = _trunc(step["text"]) summary = ( f'
' f'💭 Thinking' f'{preview}' f'
' ) body = ( f'
' f'
' f'{step["text"]}
' f'
' ) st.markdown(_details_wrap(summary, body, is_recent), unsafe_allow_html=True) def _render_tool_round(step: dict, is_recent: bool = True) -> None: call = step["call"] result = step["result"] pending = result is None if pending: bg, border_color, accent, icon = WARN_BG, WARN_BORDER, AMBER, "🔄" else: bg, border_color, accent, icon = BG_MUTED, BORDER, GREEN, "✅" name = call["name"] category = _tool_category(name) badge_html = source_badge(category) args_str = _fmt_args(call.get("args") or {}) snippet_preview = _trunc(result["snippet"] if result else "") if result else "" # Summary row (shown when collapsed) status_text = f'running…' if pending else ( f'{snippet_preview}' if snippet_preview else "" ) summary = ( f'
' f'
' f'{icon} ' f'{name}' f'' f'{status_text}' f'
' f'{badge_html}' f'
' ) # Full body (shown when open) args_html = ( f'
' f'{args_str}
' ) if args_str else "" if result: snippet = result.get("snippet") or "(empty)" details = _details_block(result.get("full") or "") result_html = ( f'
' f'
↩ Result
' f'
{snippet}
' f'{details}' f'
' ) else: result_html = ( f'
Waiting for result…
' ) body = ( f'
' f'
' f'
' f'{icon}' f'{name}' f'
' f'{badge_html}' f'
' f'{args_html}' f'{result_html}' f'
' ) st.markdown(_details_wrap(summary, body, is_recent), unsafe_allow_html=True) def _render_synthesis(step: dict) -> None: if step.get("error"): icon, label, color = "❌", f"Synthesis failed: {step['error']}", RED elif step["status"] == "in_progress": icon, label, color = "🔄", "Synthesizing brief…", AMBER else: icon, label, color = "✅", "Brief synthesized", GREEN st.markdown( f'
' f'
' f'{icon} {label}
', unsafe_allow_html=True, ) # ────────────────────────────────────────────────────────────────────────────── # Public API # ────────────────────────────────────────────────────────────────────────────── def render_trace_body(trace: list[dict], recent_n: int = 2) -> None: """Render every step of the trace. Older steps are collapsed, last recent_n are expanded.""" if not trace: st.caption("No reasoning steps yet.") return paired = _pair_steps(trace) n = len(paired) for i, step in enumerate(paired): is_recent = i >= n - recent_n kind = step["kind"] if kind == "assistant_text": _render_assistant_text(step, is_recent=is_recent) elif kind == "tool_round": _render_tool_round(step, is_recent=is_recent) elif kind == "synthesis": _render_synthesis(step) def render(trace: list[dict]) -> None: """Render the reasoning expander on the Verdict tab (collapsed by default).""" if not trace: return n_rounds = sum(1 for s in trace if s["kind"] == "tool_call") label = f"🧠 Reasoning trace — {n_rounds} tool round{'s' if n_rounds != 1 else ''}, {len(trace)} steps" with st.expander(label, expanded=False): render_trace_body(trace)