1401 lines
50 KiB
Python
1401 lines
50 KiB
Python
"""
|
||
ui.py — AIPA Terminal Interface
|
||
Authored by: Iris, Director of Interface & Experience
|
||
|
||
Visual language:
|
||
Miranda — cyan primary orchestrator, delivers to Principal
|
||
Vera — yellow auditor, independent track
|
||
Atlas — green research
|
||
Cole — blue operations
|
||
Clio — magenta analysis
|
||
Evelyn — white personnel & systems
|
||
Iris — bright_cyan interface
|
||
System — dim white infrastructure messages
|
||
Error — red failures and blockers
|
||
Warning — orange flags and cautions
|
||
|
||
Panels carry the agent name as their title. The Principal never has to read
|
||
a header to know who is speaking — color and panel style make it immediate.
|
||
"""
|
||
|
||
from contextlib import contextmanager
|
||
from datetime import datetime
|
||
import json as _json
|
||
|
||
try:
|
||
from prompt_toolkit import PromptSession as _PromptSession
|
||
from prompt_toolkit.completion import Completer as _PTCompleter, Completion as _PTCompletion
|
||
from prompt_toolkit.formatted_text import FormattedText as _FormattedText
|
||
from prompt_toolkit.history import InMemoryHistory as _InMemoryHistory
|
||
from prompt_toolkit.shortcuts import CompleteStyle as _CompleteStyle, radiolist_dialog as _radiolist_dialog
|
||
from prompt_toolkit.styles import Style as _PTStyle
|
||
_HAS_PROMPT_TOOLKIT = True
|
||
except ImportError:
|
||
_HAS_PROMPT_TOOLKIT = False
|
||
|
||
import config as _config
|
||
|
||
from rich.console import Console, Group
|
||
from rich.columns import Columns
|
||
from rich.live import Live
|
||
from rich.markdown import Markdown
|
||
from rich.panel import Panel
|
||
from rich.progress import (
|
||
BarColumn,
|
||
Progress,
|
||
SpinnerColumn,
|
||
TaskProgressColumn,
|
||
TextColumn,
|
||
TimeElapsedColumn,
|
||
)
|
||
from rich.rule import Rule
|
||
from rich.table import Table
|
||
from rich.text import Text
|
||
from rich.theme import Theme
|
||
from rich import box
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Console setup
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def _build_theme() -> Theme:
|
||
"""Build the Rich theme dynamically from agents.yaml color definitions."""
|
||
theme_dict: dict[str, str] = {
|
||
# System styles — not agent-specific
|
||
"system": "dim white",
|
||
"error": "bold red",
|
||
"warning": "bold yellow",
|
||
"task_id": "dim cyan",
|
||
"label": "bold white",
|
||
"muted": "dim",
|
||
"pass": "bold green",
|
||
"flag": "bold yellow",
|
||
"reject": "bold red",
|
||
}
|
||
# Agent name aliases — one entry per agent in agents.yaml
|
||
for name in _config.AGENT_CONFIGS:
|
||
theme_dict[name] = f"bold {_config.agent_color(name)}"
|
||
# Role aliases — stable names regardless of which agent holds the role
|
||
for role, agent in (
|
||
("orchestrator", _config.ORCHESTRATOR_AGENT),
|
||
("auditor", _config.AUDITOR_AGENT),
|
||
("recruiter", _config.RECRUITER_AGENT),
|
||
):
|
||
if agent:
|
||
theme_dict[role] = f"bold {_config.agent_color(agent)}"
|
||
return Theme(theme_dict)
|
||
|
||
|
||
console = Console(theme=_build_theme(), highlight=False)
|
||
|
||
_deliverable_streamed: bool = False # set when a deliverable/agent response was streamed
|
||
_audit_streamed: bool = False # set when an audit response was streamed
|
||
|
||
# Thinking history — most recent appended last, capped at _THINKING_MAX entries.
|
||
# Each entry: {agent, content, n_chars, n_lines, timestamp, index}
|
||
_thinking_history: list[dict] = []
|
||
_THINKING_MAX = 20
|
||
|
||
# Lead output history — stores completed lead outputs, accessible via /leads.
|
||
# Each entry: {agent, task_id, content, n_chars, n_lines, timestamp, index}
|
||
_lead_output_history: list[dict] = []
|
||
_LEAD_OUTPUT_MAX = 50
|
||
|
||
|
||
def _record_thinking(agent_name: str, content: str) -> None:
|
||
"""Append a thinking record to the history, trimming to _THINKING_MAX."""
|
||
if not content:
|
||
return
|
||
_thinking_history.append({
|
||
"agent": agent_name,
|
||
"content": content,
|
||
"n_chars": len(content),
|
||
"n_lines": content.count("\n") + 1,
|
||
"timestamp": datetime.now().strftime("%H:%M"),
|
||
"index": len(_thinking_history) + 1,
|
||
})
|
||
if len(_thinking_history) > _THINKING_MAX:
|
||
del _thinking_history[:-_THINKING_MAX]
|
||
|
||
|
||
def record_lead_output(agent_name: str, task_id: str, content: str) -> None:
|
||
"""Store a completed lead's full output in the reviewable history."""
|
||
if not content:
|
||
return
|
||
_lead_output_history.append({
|
||
"agent": agent_name,
|
||
"task_id": task_id,
|
||
"content": content,
|
||
"n_chars": len(content),
|
||
"n_lines": content.count("\n") + 1,
|
||
"timestamp": datetime.now().strftime("%H:%M"),
|
||
"index": len(_lead_output_history) + 1,
|
||
})
|
||
if len(_lead_output_history) > _LEAD_OUTPUT_MAX:
|
||
del _lead_output_history[:-_LEAD_OUTPUT_MAX]
|
||
|
||
|
||
def print_lead_dispatch(agent_name: str, task_id: str) -> None:
|
||
"""Print a one-liner when a lead is dispatched, before its Live panel opens."""
|
||
color = _agent_color(agent_name)
|
||
title = _agent_display_title(agent_name)
|
||
console.print(Text.assemble(
|
||
(" ▶ ", "dim"),
|
||
(f"{title} [{task_id}]", f"dim {color}"),
|
||
(" dispatched", "dim"),
|
||
))
|
||
|
||
|
||
def print_lead_completion(agent_name: str, task_id: str, n_chars: int, error: bool = False) -> None:
|
||
"""Print a persistent one-liner after a lead's transient panel closes — like the thinking summary line."""
|
||
color = _agent_color(agent_name)
|
||
title = _agent_display_title(agent_name)
|
||
if error:
|
||
console.print(Text.assemble(
|
||
(" ▸ ", "dim"),
|
||
(f"{title} [{task_id}]", f"dim {color}"),
|
||
(" error", "bold red"),
|
||
))
|
||
else:
|
||
console.print(Text.assemble(
|
||
(" ▸ ", "dim"),
|
||
(f"{title} [{task_id}]", f"dim {color}"),
|
||
(" complete ", "dim"),
|
||
(f"[{n_chars:,} chars]", "dim"),
|
||
(" /leads to review", "dim italic"),
|
||
))
|
||
|
||
# Slash-command completions: list of (command_name, description) pairs.
|
||
_completions: list[tuple[str, str]] = []
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Inline slash-command completion (prompt_toolkit)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
if _HAS_PROMPT_TOOLKIT:
|
||
class _SlashCompleter(_PTCompleter):
|
||
"""
|
||
Yields up to 5 matching slash commands as the user types.
|
||
Only activates when the current input starts with '/'.
|
||
Each completion shows the command name and its description.
|
||
"""
|
||
_MAX = 5
|
||
|
||
def get_completions(self, document, complete_event):
|
||
text = document.text_before_cursor
|
||
if not text.startswith("/"):
|
||
return
|
||
count = 0
|
||
for word, desc in _completions:
|
||
if word.startswith(text):
|
||
yield _PTCompletion(
|
||
word,
|
||
start_position=-len(text),
|
||
display=word,
|
||
display_meta=desc,
|
||
)
|
||
count += 1
|
||
if count >= self._MAX:
|
||
break
|
||
|
||
_PT_STYLE = _PTStyle.from_dict({
|
||
# Prompt text
|
||
"prompt": "bold #ffffff",
|
||
# Completion dropdown — unselected
|
||
"completion-menu.completion": "bg:#1e1e1e #aaaaaa",
|
||
"completion-menu.meta.completion": "bg:#1e1e1e #555555",
|
||
# Completion dropdown — selected row
|
||
"completion-menu.completion.current": "bg:#005f87 bold #ffffff",
|
||
"completion-menu.meta.completion.current": "bg:#005f87 #bbbbbb",
|
||
# Scrollbar
|
||
"scrollbar.background": "bg:#1e1e1e",
|
||
"scrollbar.button": "bg:#4e4e4e",
|
||
})
|
||
|
||
_pt_session: _PromptSession | None = _PromptSession(
|
||
history=_InMemoryHistory(),
|
||
completer=_SlashCompleter(),
|
||
complete_while_typing=True, # popup appears as the user types '/'
|
||
complete_style=_CompleteStyle.COLUMN,
|
||
style=_PT_STYLE,
|
||
)
|
||
else:
|
||
_pt_session = None
|
||
|
||
|
||
def set_completions(entries: list[tuple[str, str]]) -> None:
|
||
"""
|
||
Register (command_name, description) pairs for inline completion.
|
||
Call once at startup after all slash commands are defined.
|
||
"""
|
||
global _completions
|
||
_completions = sorted(entries, key=lambda e: e[0])
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Agent visual config
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def _agent_color(name: str) -> str:
|
||
return _config.agent_color(name.lower())
|
||
|
||
|
||
def _agent_display_title(name: str) -> str:
|
||
"""Return 'Name — Title' using the title from agents.yaml."""
|
||
title = _config.agent_title(name)
|
||
return f"{name.capitalize()} — {title}" if title else name.capitalize()
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Session chrome
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def print_session_header():
|
||
"""Opening banner for a new Principal session."""
|
||
cos_name = _config.ORCHESTRATOR_AGENT.capitalize()
|
||
cos_title = _config.agent_title(_config.ORCHESTRATOR_AGENT)
|
||
aud_name = _config.AUDITOR_AGENT.capitalize()
|
||
app_name = _config.APP_NAME
|
||
|
||
console.print()
|
||
console.rule(f"[bold cyan]{app_name}[/bold cyan] [dim]Autonomous Intelligence Principal Architecture[/dim]", style="cyan")
|
||
console.print()
|
||
|
||
table = Table(box=box.SIMPLE, show_header=False, padding=(0, 2))
|
||
table.add_column(style="dim")
|
||
table.add_column(style="white")
|
||
table.add_row("Session opened", datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
|
||
table.add_row("Orchestrator", f"{cos_name} — {cos_title}")
|
||
table.add_row("Auditor", f"{aud_name} (independent track)")
|
||
table.add_row("Commands", "/help for available commands")
|
||
console.print(table)
|
||
console.print()
|
||
|
||
|
||
def print_session_footer(task_count: int):
|
||
"""Closing rule when the session ends."""
|
||
console.print()
|
||
console.rule(
|
||
f"[dim]Session closed — {task_count} directive{'s' if task_count != 1 else ''} processed[/dim]",
|
||
style="dim"
|
||
)
|
||
console.print()
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Input prompt
|
||
# ---------------------------------------------------------------------------
|
||
|
||
async def get_directive() -> str:
|
||
"""
|
||
Styled input prompt for the Principal.
|
||
|
||
When prompt_toolkit is available (the default):
|
||
- An inline vertical dropdown appears as soon as the user types '/'.
|
||
- Up to 5 matching commands are shown with their descriptions.
|
||
- Arrow keys navigate the list; Enter or Tab accepts a selection.
|
||
- Up/Down arrow keys also navigate input history within the session.
|
||
|
||
Uses prompt_async() so it cooperates with the running asyncio event loop
|
||
instead of trying to nest a second one (which raises RuntimeError).
|
||
|
||
Falls back to plain input() if prompt_toolkit is not installed.
|
||
"""
|
||
console.print()
|
||
if _pt_session is not None:
|
||
return (await _pt_session.prompt_async(
|
||
_FormattedText([("class:prompt", "Principal › ")]),
|
||
)).strip()
|
||
# Fallback: plain styled prompt with no completion.
|
||
console.print("[bold white]Principal ›[/bold white] ", end="")
|
||
if hasattr(console.file, "flush"):
|
||
console.file.flush()
|
||
return input().strip()
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# System / status messages
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def print_system(message: str):
|
||
console.print(f" [system]⬡ {message}[/system]")
|
||
|
||
|
||
def print_error(message: str):
|
||
console.print(f"\n [error]✗ {message}[/error]\n")
|
||
|
||
|
||
def print_warning(message: str):
|
||
console.print(f" [warning]⚠ {message}[/warning]")
|
||
|
||
|
||
def print_task_received(task_id: str, directive_preview: str):
|
||
console.print()
|
||
console.print(
|
||
f" [task_id]{task_id}[/task_id] [dim]Directive received:[/dim] "
|
||
f"[white]{directive_preview[:100]}{'…' if len(directive_preview) > 100 else ''}[/white]"
|
||
)
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Lead dispatch progress
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def _brief_preview(brief: str, max_len: int = 72) -> str:
|
||
"""Return a short preview of a task brief, skipping the header lines."""
|
||
lines = [l.strip() for l in brief.splitlines() if l.strip()]
|
||
content = [
|
||
l for l in lines
|
||
if not l.upper().startswith("TASK BRIEF") and not l.upper().startswith("TO:")
|
||
]
|
||
text = " ".join(content)
|
||
return (text[:max_len] + "…") if len(text) > max_len else text
|
||
|
||
|
||
def print_task_dispatch_plan(tasks: list):
|
||
"""
|
||
Show the task assignments before leads are dispatched.
|
||
Each row: task_id | lead | brief preview.
|
||
"""
|
||
console.print()
|
||
console.rule("[dim]Lead dispatch[/dim]", style="dim")
|
||
table = Table(box=box.SIMPLE, show_header=False, padding=(0, 2), expand=False)
|
||
table.add_column(style="task_id", no_wrap=True)
|
||
table.add_column(no_wrap=True)
|
||
table.add_column(style="dim")
|
||
for t in tasks:
|
||
color = _agent_color(t.assigned_to)
|
||
table.add_row(
|
||
t.task_id,
|
||
f"[{color}]{_agent_display_title(t.assigned_to)}[/{color}]",
|
||
_brief_preview(t.brief),
|
||
)
|
||
console.print(table)
|
||
|
||
|
||
def print_direct_response_notice():
|
||
"""Shown when Miranda handles a directive without dispatching leads."""
|
||
console.print(
|
||
" [dim]Miranda responded directly — no lead dispatch required.[/dim]"
|
||
)
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Agent output panels
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def print_agent_panel(agent_name: str, content: str, task_id: str = ""):
|
||
"""
|
||
Render an agent's output in a styled panel.
|
||
Content is treated as Markdown.
|
||
Skips printing if stream_agent_output already rendered this response.
|
||
"""
|
||
global _deliverable_streamed
|
||
if _deliverable_streamed:
|
||
_deliverable_streamed = False
|
||
return
|
||
|
||
color = _agent_color(agent_name)
|
||
title_text = _agent_display_title(agent_name)
|
||
subtitle = f"[dim]{task_id}[/dim]" if task_id else ""
|
||
|
||
panel = Panel(
|
||
Markdown(content),
|
||
title=f"[{color}] {title_text} [/{color}]",
|
||
subtitle=subtitle,
|
||
border_style=color,
|
||
padding=(1, 2),
|
||
box=box.ROUNDED,
|
||
)
|
||
console.print()
|
||
console.print(panel)
|
||
|
||
|
||
def print_deliverable(content: str, task_id: str = ""):
|
||
"""
|
||
Orchestrator's synthesized deliverable.
|
||
Skips printing if stream_agent_output already rendered this response.
|
||
"""
|
||
global _deliverable_streamed
|
||
if _deliverable_streamed:
|
||
_deliverable_streamed = False
|
||
return
|
||
_print_deliverable_panel(content, task_id)
|
||
|
||
|
||
def _print_deliverable_panel(content: str, task_id: str = ""):
|
||
"""Internal: always render the deliverable panel unconditionally."""
|
||
agent_name = _config.ORCHESTRATOR_AGENT
|
||
color = _agent_color(agent_name)
|
||
title_text = _agent_display_title(agent_name)
|
||
subtitle = f"[dim]{task_id}[/dim]" if task_id else ""
|
||
|
||
panel = Panel(
|
||
Markdown(content),
|
||
title=f"[{color}] {title_text} [/{color}]",
|
||
subtitle=subtitle,
|
||
border_style=color,
|
||
padding=(1, 2),
|
||
box=box.ROUNDED,
|
||
)
|
||
console.print()
|
||
console.print(panel)
|
||
|
||
|
||
def print_audit_memo(content: str, task_id: str = ""):
|
||
"""
|
||
Auditor's memo. Verdict line drives the border colour.
|
||
PASS = green, PASS WITH NOTES = yellow, FLAG = yellow, REJECT = red.
|
||
Skips printing if stream_agent_output already rendered this response.
|
||
"""
|
||
global _audit_streamed
|
||
if _audit_streamed:
|
||
_audit_streamed = False
|
||
return
|
||
_print_audit_memo_panel(content, task_id)
|
||
|
||
|
||
def _print_audit_memo_panel(content: str, task_id: str = ""):
|
||
"""Internal: always render the audit memo panel with verdict-based border."""
|
||
aud_name = _config.AUDITOR_AGENT
|
||
aud_title = _config.agent_title(aud_name)
|
||
aud_color = _agent_color(aud_name)
|
||
|
||
verdict_style = "dim"
|
||
for line in content.splitlines():
|
||
upper = line.upper()
|
||
if "VERDICT:" in upper:
|
||
if "REJECT" in upper:
|
||
verdict_style = "reject"
|
||
elif "FLAG" in upper:
|
||
verdict_style = "flag"
|
||
elif "PASS WITH NOTES" in upper:
|
||
verdict_style = "warning"
|
||
elif "PASS" in upper:
|
||
verdict_style = "pass"
|
||
break
|
||
|
||
panel = Panel(
|
||
Markdown(content),
|
||
title=f"[{aud_color}] {aud_name.capitalize()} — {aud_title} [/{aud_color}]",
|
||
subtitle=f"[dim]{task_id}[/dim]" if task_id else "",
|
||
border_style=verdict_style if verdict_style != "dim" else aud_color,
|
||
padding=(1, 2),
|
||
box=box.ROUNDED,
|
||
)
|
||
console.print()
|
||
console.print(panel)
|
||
|
||
|
||
def print_lead_output(agent_name: str, content: str, task_id: str = ""):
|
||
"""A lead's raw output (shown only in DEBUG mode)."""
|
||
print_agent_panel(agent_name, content, task_id)
|
||
|
||
|
||
def print_summary(content: str, source_agent: str = ""):
|
||
"""Render Tillie's TLDR summary in a labelled panel.
|
||
source_agent: the agent whose output is being summarized (e.g. 'miranda', 'vera').
|
||
"""
|
||
summarizer = _config.SUMMARIZER_AGENT
|
||
color = _agent_color(summarizer) if summarizer else "white"
|
||
if source_agent:
|
||
title_text = f"Tillie — TLDR: {source_agent.capitalize()}"
|
||
else:
|
||
title_text = "Tillie — TLDR Summarizer"
|
||
panel = Panel(
|
||
Markdown(content),
|
||
title=f"[{color}] {title_text} [/{color}]",
|
||
border_style=color,
|
||
padding=(1, 2),
|
||
box=box.ROUNDED,
|
||
)
|
||
console.print()
|
||
console.print(panel)
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Standing Brief renderer
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def print_standing_brief(raw_md: str):
|
||
"""
|
||
Render the standing brief as a navigable terminal layout.
|
||
Sections are separated by rules; tables are preserved where present.
|
||
"""
|
||
console.print()
|
||
console.rule("[bold cyan]Standing Brief[/bold cyan]", style="cyan")
|
||
console.print()
|
||
console.print(Markdown(raw_md))
|
||
console.print()
|
||
console.rule(style="dim")
|
||
console.print()
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Synthesis progress (Miranda)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
@contextmanager
|
||
def synthesis_progress():
|
||
"""Spinner shown while Miranda synthesizes lead outputs."""
|
||
with Progress(
|
||
SpinnerColumn(style="cyan"),
|
||
TextColumn("[cyan]Miranda synthesizing…[/cyan]"),
|
||
TimeElapsedColumn(),
|
||
console=console,
|
||
transient=True,
|
||
) as progress:
|
||
progress.add_task("", total=None)
|
||
yield
|
||
|
||
|
||
@contextmanager
|
||
def audit_progress():
|
||
"""Spinner shown while Vera audits."""
|
||
with Progress(
|
||
SpinnerColumn(style="yellow"),
|
||
TextColumn("[yellow]Vera auditing…[/yellow]"),
|
||
TimeElapsedColumn(),
|
||
console=console,
|
||
transient=True,
|
||
) as progress:
|
||
progress.add_task("", total=None)
|
||
yield
|
||
|
||
|
||
@contextmanager
|
||
def brief_update_progress():
|
||
"""Spinner shown while Miranda updates the standing brief."""
|
||
with Progress(
|
||
SpinnerColumn(style="dim"),
|
||
TextColumn("[dim]Updating standing brief…[/dim]"),
|
||
console=console,
|
||
transient=True,
|
||
) as progress:
|
||
progress.add_task("", total=None)
|
||
yield
|
||
|
||
|
||
@contextmanager
|
||
def summary_progress():
|
||
"""Spinner shown while Tillie produces a TLDR summary."""
|
||
summarizer = _config.SUMMARIZER_AGENT
|
||
color = _agent_color(summarizer) if summarizer else "white"
|
||
with Progress(
|
||
SpinnerColumn(style=color),
|
||
TextColumn(f"[{color}]Tillie summarizing…[/{color}]"),
|
||
TimeElapsedColumn(),
|
||
console=console,
|
||
transient=True,
|
||
) as progress:
|
||
progress.add_task("", total=None)
|
||
yield
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Streaming output with thinking support
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def _thinking_panel_title(agent_name: str, mode: str) -> str:
|
||
"""Return the thinking panel title, tinted by agent identity."""
|
||
if mode == "audit":
|
||
color = _agent_color(agent_name)
|
||
return f"[{color}]Vera thinking…[/{color}]"
|
||
color = _agent_color(agent_name)
|
||
return f"[{color} dim]Thinking…[/{color} dim]"
|
||
|
||
|
||
def _build_thinking_panel(thinking_text: str, agent_name: str, mode: str,
|
||
visible_lines: int = 8) -> object:
|
||
"""
|
||
Build the live thinking panel for Phase 1 streaming.
|
||
|
||
Shows the last `visible_lines` lines of thinking so the panel never
|
||
grows beyond a predictable height (safe with vertical_overflow="crop").
|
||
An overflow indicator shows how many earlier lines have scrolled off.
|
||
"""
|
||
if not thinking_text:
|
||
return Text(
|
||
f" [{_agent_color(agent_name)} dim]{_agent_display_title(agent_name)} thinking…"
|
||
f"[/{_agent_color(agent_name)} dim]",
|
||
style="dim",
|
||
)
|
||
|
||
lines = thinking_text.splitlines()
|
||
hidden = max(0, len(lines) - visible_lines)
|
||
visible = lines[-visible_lines:] if hidden else lines
|
||
|
||
items = []
|
||
if hidden:
|
||
items.append(Text(f" +{hidden} line{'s' if hidden != 1 else ''} earlier\n",
|
||
style="dim italic"))
|
||
items.append(Text("\n".join(visible), style="dim"))
|
||
content = Group(*items) if len(items) > 1 else items[0]
|
||
|
||
return Panel(
|
||
content,
|
||
title=_thinking_panel_title(agent_name, mode),
|
||
border_style="dim",
|
||
padding=(0, 1),
|
||
box=box.SIMPLE,
|
||
)
|
||
|
||
|
||
async def stream_agent_output(
|
||
agent_name: str,
|
||
stream_gen,
|
||
mode: str = "deliverable",
|
||
) -> str:
|
||
"""
|
||
Consume an async generator of ('thinking'|'content', text) chunks and
|
||
render them live in the terminal.
|
||
|
||
Two-phase rendering
|
||
-------------------
|
||
Phase 1 — Thinking (transient, cropped to terminal height):
|
||
A compact Live panel streams thinking text showing the last 8 lines.
|
||
It is transient — it clears from the screen when Phase 1 ends.
|
||
A persistent one-line summary is then printed to the scroll buffer:
|
||
"▸ Agent — thinking complete [N chars · L lines] /thinking to expand"
|
||
|
||
Phase 2 — Content (non-transient, persists in scroll buffer):
|
||
A separate Live display streams the content and renders the final
|
||
Markdown panel inside the Live block before exiting (no flash).
|
||
|
||
This keeps thinking and content in separate, non-competing display
|
||
regions. Thinking never pushes content off-screen, and the summary
|
||
line is always visible in the scroll buffer.
|
||
|
||
mode values
|
||
-----------
|
||
"deliverable" Normal agent response.
|
||
"audit" Like deliverable but with Vera's identity and verdict border.
|
||
"background" Housekeeping — compact transient spinner only, no panel.
|
||
|
||
In all modes the thinking content is recorded via _record_thinking() so
|
||
the /thinking picker has a full history.
|
||
Returns the full content string.
|
||
"""
|
||
global _deliverable_streamed, _audit_streamed
|
||
|
||
color = _agent_color(agent_name)
|
||
title_text = _agent_display_title(agent_name)
|
||
|
||
# ── Lead mode ────────────────────────────────────────────────────────────
|
||
# Transient height-capped Live panel — mirrors Phase 1 thinking display.
|
||
# Content streams line-by-line so the panel updates visibly.
|
||
# The caller (dispatch_to_lead) prints the one-liner and stores history.
|
||
# Deliberately does NOT set _deliverable_streamed.
|
||
if mode == "lead":
|
||
import asyncio as _asyncio
|
||
|
||
content_parts: list[str] = []
|
||
async for chunk_type, text in stream_gen:
|
||
if chunk_type == "content":
|
||
content_parts.append(text)
|
||
full_content = "".join(content_parts)
|
||
|
||
visible_lines = 8
|
||
|
||
def _lead_panel(accumulated: str) -> object:
|
||
if not accumulated:
|
||
return Text(
|
||
f" [{color} dim]{title_text} working…[/{color} dim]",
|
||
style="dim",
|
||
)
|
||
lines = accumulated.splitlines()
|
||
hidden = max(0, len(lines) - visible_lines)
|
||
visible = lines[-visible_lines:] if hidden else lines
|
||
items = []
|
||
if hidden:
|
||
items.append(Text(
|
||
f" +{hidden} line{'s' if hidden != 1 else ''} earlier\n",
|
||
style="dim italic",
|
||
))
|
||
items.append(Text("\n".join(visible), style=f"dim {color}"))
|
||
content_renderable = Group(*items) if len(items) > 1 else items[0]
|
||
return Panel(
|
||
content_renderable,
|
||
title=f"[{color} dim]{title_text}…[/{color} dim]",
|
||
border_style="dim",
|
||
padding=(0, 1),
|
||
box=box.SIMPLE,
|
||
)
|
||
|
||
with Live(
|
||
_lead_panel(""),
|
||
console=console,
|
||
refresh_per_second=15,
|
||
transient=True,
|
||
vertical_overflow="crop",
|
||
) as live:
|
||
lines = full_content.splitlines(keepends=True)
|
||
accumulated = ""
|
||
for line in lines:
|
||
accumulated += line
|
||
live.update(_lead_panel(accumulated))
|
||
await _asyncio.sleep(0) # yield to event loop so Rich can repaint
|
||
|
||
return full_content
|
||
|
||
# ── Background mode ──────────────────────────────────────────────────────
|
||
if mode == "background":
|
||
thinking_parts: list[str] = []
|
||
content_parts: list[str] = []
|
||
with Progress(
|
||
SpinnerColumn(style="dim"),
|
||
TextColumn("[dim]Updating standing brief…[/dim]"),
|
||
TimeElapsedColumn(),
|
||
console=console,
|
||
transient=True,
|
||
) as prog:
|
||
prog.add_task("", total=None)
|
||
async for chunk_type, text in stream_gen:
|
||
if chunk_type == "thinking":
|
||
thinking_parts.append(text)
|
||
elif chunk_type == "content":
|
||
content_parts.append(text)
|
||
_record_thinking(agent_name, "".join(thinking_parts))
|
||
_deliverable_streamed = True
|
||
return "".join(content_parts)
|
||
|
||
# ── Deliverable / audit modes ─────────────────────────────────────────────
|
||
thinking_parts: list[str] = []
|
||
content_parts: list[str] = []
|
||
|
||
def _build_final_panel(content_text: str) -> Panel:
|
||
if mode == "audit":
|
||
aud_name = _config.AUDITOR_AGENT
|
||
aud_title = _config.agent_title(aud_name)
|
||
aud_color = _agent_color(aud_name)
|
||
verdict_style = "dim"
|
||
for line in content_text.splitlines():
|
||
upper = line.upper()
|
||
if "VERDICT:" in upper:
|
||
if "REJECT" in upper: verdict_style = "reject"
|
||
elif "FLAG" in upper: verdict_style = "flag"
|
||
elif "PASS WITH NOTES" in upper: verdict_style = "warning"
|
||
elif "PASS" in upper: verdict_style = "pass"
|
||
break
|
||
return Panel(
|
||
Markdown(content_text),
|
||
title=f"[{aud_color}] {aud_name.capitalize()} — {aud_title} [/{aud_color}]",
|
||
border_style=verdict_style if verdict_style != "dim" else aud_color,
|
||
padding=(1, 2), box=box.ROUNDED,
|
||
)
|
||
return Panel(
|
||
Markdown(content_text),
|
||
title=f"[{color}] {title_text} [/{color}]",
|
||
border_style=color,
|
||
padding=(1, 2), box=box.ROUNDED,
|
||
)
|
||
|
||
console.print()
|
||
|
||
# ── Phase 1: Thinking — transient, height-capped Live display ────────────
|
||
# We use manual __anext__() instead of `async for` so that breaking out of
|
||
# the loop does NOT call aclose() on the generator. The same iterator is
|
||
# then reused in Phase 2 to continue from where we left off.
|
||
aiter = stream_gen.__aiter__()
|
||
exhausted = False
|
||
|
||
with Live(
|
||
_build_thinking_panel("", agent_name, mode),
|
||
console=console,
|
||
refresh_per_second=15,
|
||
transient=True, # clears from screen when the block exits
|
||
vertical_overflow="crop", # never grows beyond terminal height
|
||
) as live:
|
||
while True:
|
||
try:
|
||
chunk_type, text = await aiter.__anext__()
|
||
except StopAsyncIteration:
|
||
exhausted = True
|
||
break
|
||
if chunk_type == "thinking":
|
||
thinking_parts.append(text)
|
||
live.update(_build_thinking_panel(
|
||
"".join(thinking_parts), agent_name, mode
|
||
))
|
||
elif chunk_type == "content":
|
||
content_parts.append(text)
|
||
break # hand off to Phase 2; generator is NOT closed
|
||
|
||
# Print a persistent collapsed summary when thinking was present.
|
||
# This line lives in the terminal's scroll buffer — always scrollable.
|
||
if thinking_parts:
|
||
full_thinking = "".join(thinking_parts)
|
||
n_chars = len(full_thinking)
|
||
n_lines = full_thinking.count("\n") + 1
|
||
id_color = _agent_color(_config.AUDITOR_AGENT) if mode == "audit" else color
|
||
console.print(Text.assemble(
|
||
(" ▸ ", "dim"),
|
||
(title_text, f"dim {id_color}"),
|
||
(" thinking complete ", "dim"),
|
||
(f"[{n_chars:,} chars · {n_lines} lines]", "dim"),
|
||
(" /thinking to expand", "dim italic"),
|
||
))
|
||
|
||
# Edge case: generator exhausted with no content (thinking-only response)
|
||
if exhausted:
|
||
_record_thinking(agent_name, "".join(thinking_parts))
|
||
if mode == "audit":
|
||
_audit_streamed = True
|
||
else:
|
||
_deliverable_streamed = True
|
||
return ""
|
||
|
||
# ── Phase 2: Content — non-transient, persists in scroll buffer ──────────
|
||
def _content_renderable(final: bool = False) -> object:
|
||
content_text = "".join(content_parts)
|
||
if not content_text:
|
||
return Text(
|
||
f" [{color} dim]{title_text} …[/{color} dim]",
|
||
style="dim",
|
||
)
|
||
if final:
|
||
return _build_final_panel(content_text)
|
||
return Panel(
|
||
Text(content_text),
|
||
title=f"[{color}] {title_text} [/{color}]",
|
||
border_style=color,
|
||
padding=(1, 2), box=box.ROUNDED,
|
||
)
|
||
|
||
with Live(
|
||
_content_renderable(),
|
||
console=console,
|
||
refresh_per_second=15,
|
||
transient=False,
|
||
vertical_overflow="visible",
|
||
) as live:
|
||
async for chunk_type, text in aiter:
|
||
if chunk_type == "content":
|
||
content_parts.append(text)
|
||
live.update(_content_renderable())
|
||
live.update(_content_renderable(final=True))
|
||
|
||
_record_thinking(agent_name, "".join(thinking_parts))
|
||
if mode == "audit":
|
||
_audit_streamed = True
|
||
else:
|
||
_deliverable_streamed = True
|
||
return "".join(content_parts)
|
||
|
||
|
||
def get_last_thinking() -> str:
|
||
"""Return the thinking text from the most recent response (empty string if none)."""
|
||
return _thinking_history[-1]["content"] if _thinking_history else ""
|
||
|
||
|
||
def get_lead_output_history() -> list[dict]:
|
||
"""Return the full lead output history list (most recent last)."""
|
||
return _lead_output_history
|
||
|
||
|
||
async def show_lead_output_picker() -> None:
|
||
"""
|
||
Browse completed lead outputs for this session.
|
||
Mirrors show_thinking_picker() in structure and interaction.
|
||
"""
|
||
if not _lead_output_history:
|
||
print_system("No lead outputs recorded this session.")
|
||
return
|
||
|
||
def _display_lead_entry(entry: dict) -> None:
|
||
color = _agent_color(entry["agent"])
|
||
title = _agent_display_title(entry["agent"])
|
||
panel = Panel(
|
||
Markdown(entry["content"]),
|
||
title=(
|
||
f"[{color}] {title} [{entry['task_id']}] "
|
||
f"#{entry['index']} "
|
||
f"[{entry['n_chars']:,} chars · {entry['n_lines']} lines] [/{color}]"
|
||
),
|
||
subtitle=f"[dim]{entry['timestamp']}[/dim]",
|
||
border_style=color,
|
||
padding=(1, 2),
|
||
box=box.ROUNDED,
|
||
)
|
||
console.print()
|
||
console.print(panel)
|
||
|
||
if len(_lead_output_history) == 1:
|
||
_display_lead_entry(_lead_output_history[0])
|
||
return
|
||
|
||
if _HAS_PROMPT_TOOLKIT:
|
||
values = [
|
||
(
|
||
entry,
|
||
(
|
||
f"#{entry['index']} "
|
||
f"{entry['agent'].capitalize()} [{entry['task_id']}] "
|
||
f"[{entry['n_chars']:,} chars · {entry['n_lines']} lines]"
|
||
f" {entry['timestamp']}"
|
||
),
|
||
)
|
||
for entry in reversed(_lead_output_history)
|
||
]
|
||
selected = await _radiolist_dialog(
|
||
title="Lead Output History",
|
||
text=f"Select a lead output to view ({len(_lead_output_history)} stored):",
|
||
values=values,
|
||
style=_PT_STYLE,
|
||
).run_async()
|
||
if selected is not None:
|
||
_display_lead_entry(selected)
|
||
else:
|
||
console.print()
|
||
console.rule("[dim]Lead Output History[/dim]", style="dim")
|
||
console.print()
|
||
for entry in reversed(_lead_output_history):
|
||
color = _agent_color(entry["agent"])
|
||
console.print(
|
||
f" [dim]{entry['index']:>2}.[/dim] "
|
||
f"[{color}]{entry['agent'].capitalize()}[/{color}] "
|
||
f"[dim]{entry['task_id']} "
|
||
f"{entry['n_chars']:,} chars · {entry['n_lines']} lines"
|
||
f" {entry['timestamp']}[/dim]"
|
||
)
|
||
console.print()
|
||
try:
|
||
choice = input(" Enter number to view (or Enter to cancel): ").strip()
|
||
if choice.isdigit():
|
||
match = next(
|
||
(e for e in _lead_output_history if e["index"] == int(choice)), None
|
||
)
|
||
if match:
|
||
_display_lead_entry(match)
|
||
else:
|
||
print_system(f"No lead output #{choice}.")
|
||
except (EOFError, KeyboardInterrupt):
|
||
pass
|
||
|
||
|
||
def _display_thinking_entry(entry: dict) -> None:
|
||
"""Render one thinking history entry in a dim panel."""
|
||
title_text = _agent_display_title(entry["agent"])
|
||
panel = Panel(
|
||
Text(entry["content"], style="dim"),
|
||
title=(
|
||
f"[dim] {title_text} — thinking "
|
||
f"#{entry['index']} "
|
||
f"[{entry['n_chars']:,} chars · {entry['n_lines']} lines] [/dim]"
|
||
),
|
||
subtitle=f"[dim]{entry['timestamp']}[/dim]",
|
||
border_style="dim",
|
||
padding=(1, 2),
|
||
box=box.ROUNDED,
|
||
)
|
||
console.print()
|
||
console.print(panel)
|
||
|
||
|
||
async def show_thinking_picker() -> None:
|
||
"""
|
||
Browse the thinking history for this session.
|
||
|
||
- 0 entries : prints a system notice.
|
||
- 1 entry : displays it directly.
|
||
- 2+ entries: opens an interactive radiolist dialog (prompt_toolkit) so the
|
||
user can navigate with arrow keys and press Enter to view an
|
||
entry. Falls back to a numbered text list + input() prompt
|
||
when prompt_toolkit is unavailable.
|
||
"""
|
||
if not _thinking_history:
|
||
print_system("No thinking output recorded — thinking mode may not be active.")
|
||
return
|
||
|
||
if len(_thinking_history) == 1:
|
||
_display_thinking_entry(_thinking_history[0])
|
||
return
|
||
|
||
if _HAS_PROMPT_TOOLKIT:
|
||
# Show most-recent-first in the dialog.
|
||
values = [
|
||
(
|
||
entry,
|
||
(
|
||
f"#{entry['index']} "
|
||
f"{entry['agent'].capitalize()} "
|
||
f"[{entry['n_chars']:,} chars · {entry['n_lines']} lines]"
|
||
f" {entry['timestamp']}"
|
||
),
|
||
)
|
||
for entry in reversed(_thinking_history)
|
||
]
|
||
selected = await _radiolist_dialog(
|
||
title="Thinking History",
|
||
text=f"Select a thinking output to view ({len(_thinking_history)} stored):",
|
||
values=values,
|
||
style=_PT_STYLE,
|
||
).run_async()
|
||
if selected is not None:
|
||
_display_thinking_entry(selected)
|
||
else:
|
||
# Fallback: numbered list + input prompt.
|
||
console.print()
|
||
console.rule("[dim]Thinking History[/dim]", style="dim")
|
||
console.print()
|
||
for entry in reversed(_thinking_history):
|
||
console.print(
|
||
f" [dim]{entry['index']:>2}.[/dim] "
|
||
f"[white]{entry['agent'].capitalize()}[/white] "
|
||
f"[dim]{entry['n_chars']:,} chars · {entry['n_lines']} lines"
|
||
f" {entry['timestamp']}[/dim]"
|
||
)
|
||
console.print()
|
||
try:
|
||
choice = input(" Enter number to view (or Enter to cancel): ").strip()
|
||
if choice.isdigit():
|
||
match = next(
|
||
(e for e in _thinking_history if e["index"] == int(choice)), None
|
||
)
|
||
if match:
|
||
_display_thinking_entry(match)
|
||
else:
|
||
print_system(f"No thinking output #{choice}.")
|
||
except (EOFError, KeyboardInterrupt):
|
||
pass
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Dry-run callout
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def print_dry_run_banner():
|
||
panel = Panel(
|
||
"[bold yellow]DRY RUN MODE[/bold yellow]\n"
|
||
"[dim]Routing and pipeline logic will execute normally.\n"
|
||
"No provider API calls will be made.[/dim]",
|
||
border_style="yellow",
|
||
box=box.ROUNDED,
|
||
padding=(0, 2),
|
||
)
|
||
console.print()
|
||
console.print(panel)
|
||
console.print()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Slash command displays
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def print_help(commands: list[tuple[str, str]]):
|
||
"""Render the slash command reference table."""
|
||
console.print()
|
||
console.rule("[bold cyan]Commands[/bold cyan]", style="cyan")
|
||
console.print()
|
||
|
||
table = Table(box=box.SIMPLE, show_header=False, padding=(0, 2), expand=False)
|
||
table.add_column(style="bold cyan", no_wrap=True)
|
||
table.add_column(style="white")
|
||
|
||
for cmd, description in commands:
|
||
table.add_row(cmd, description)
|
||
|
||
console.print(table)
|
||
console.print()
|
||
|
||
|
||
def print_status(
|
||
session_start: datetime,
|
||
directive_count: int,
|
||
session_log: list[str],
|
||
auto_audit: bool,
|
||
debug: bool,
|
||
dry_run: bool,
|
||
):
|
||
"""Render current session status."""
|
||
console.print()
|
||
console.rule("[bold cyan]Session Status[/bold cyan]", style="cyan")
|
||
console.print()
|
||
|
||
elapsed = datetime.now() - session_start
|
||
hours, remainder = divmod(int(elapsed.total_seconds()), 3600)
|
||
minutes, seconds = divmod(remainder, 60)
|
||
elapsed_str = f"{hours}h {minutes}m {seconds}s" if hours else f"{minutes}m {seconds}s"
|
||
|
||
global_provider = _config.ACTIVE_PROVIDER
|
||
provider_display = (
|
||
global_provider if global_provider != "none"
|
||
else "[dim]none — set default_provider in config/agents.yaml[/dim]"
|
||
)
|
||
|
||
table = Table(box=box.SIMPLE, show_header=False, padding=(0, 2))
|
||
table.add_column(style="dim", no_wrap=True)
|
||
table.add_column(style="white")
|
||
table.add_row("Session started", session_start.strftime("%Y-%m-%d %H:%M:%S"))
|
||
table.add_row("Elapsed", elapsed_str)
|
||
table.add_row("Directives", str(directive_count))
|
||
table.add_row("Global provider", provider_display)
|
||
table.add_row("Auto-audit", "[green]on[/green]" if auto_audit else "[dim]off[/dim]")
|
||
table.add_row("Debug output", "[yellow]on[/yellow]" if debug else "[dim]off[/dim]")
|
||
table.add_row("Dry run", "[yellow]on[/yellow]" if dry_run else "[dim]off[/dim]")
|
||
console.print(table)
|
||
|
||
if session_log:
|
||
console.print()
|
||
console.rule("[dim]Session log[/dim]", style="dim")
|
||
console.print()
|
||
for entry in session_log:
|
||
console.print(f" [dim]·[/dim] [muted]{entry}[/muted]")
|
||
|
||
console.print()
|
||
|
||
|
||
def print_agents(agent_configs: dict, active_provider: str):
|
||
"""Render the named agent roster with per-agent provider resolution."""
|
||
console.print()
|
||
console.rule("[bold cyan]Agent Roster[/bold cyan]", style="cyan")
|
||
console.print()
|
||
|
||
table = Table(box=box.SIMPLE, show_header=True, padding=(0, 2), expand=False)
|
||
table.add_column("Agent", style="bold white", no_wrap=True)
|
||
table.add_column("Role", style="dim", no_wrap=True)
|
||
table.add_column("Provider", style="white", no_wrap=True)
|
||
table.add_column("Model", style="cyan")
|
||
table.add_column("Temp", style="dim", justify="right")
|
||
table.add_column("Tokens", style="dim", justify="right")
|
||
|
||
for name, cfg in agent_configs.items():
|
||
color = _agent_color(name)
|
||
label = _config.agent_title(name)
|
||
provider = _config.agent_provider_name(name)
|
||
model = _config.agent_model(name) or "[dim]unset[/dim]"
|
||
|
||
provider_display = (
|
||
f"[dim]{provider}[/dim]" if provider == "none"
|
||
else provider
|
||
)
|
||
|
||
table.add_row(
|
||
f"[{color}]{name.capitalize()}[/{color}]",
|
||
label,
|
||
provider_display,
|
||
model,
|
||
str(cfg["temperature"]),
|
||
str(cfg["max_tokens"]),
|
||
)
|
||
|
||
console.print(table)
|
||
if active_provider == "none":
|
||
console.print(" [dim]Global fallback: none — set default_provider in config/agents.yaml[/dim]")
|
||
else:
|
||
console.print(f" [dim]Global fallback provider:[/dim] [white]{active_provider}[/white]")
|
||
console.print()
|
||
|
||
|
||
def print_history(session_log: list[str]):
|
||
"""Render the session log for this session."""
|
||
console.print()
|
||
console.rule("[bold cyan]Session History[/bold cyan]", style="cyan")
|
||
console.print()
|
||
|
||
if not session_log:
|
||
console.print(" [dim]No directives processed yet.[/dim]")
|
||
else:
|
||
for i, entry in enumerate(session_log, 1):
|
||
console.print(f" [dim]{i:>2}.[/dim] [muted]{entry}[/muted]")
|
||
|
||
console.print()
|
||
|
||
|
||
def print_tasks(tasks: list, session_id: str = ""):
|
||
"""Render the task roster for the current session."""
|
||
console.print()
|
||
title = f"[bold cyan]Tasks[/bold cyan]"
|
||
if session_id:
|
||
title += f" [dim]{session_id}[/dim]"
|
||
console.rule(title, style="cyan")
|
||
console.print()
|
||
|
||
if not tasks:
|
||
console.print(" [dim]No tasks dispatched this session.[/dim]")
|
||
console.print()
|
||
return
|
||
|
||
STATUS_STYLE = {
|
||
"pending": "[dim]pending[/dim]",
|
||
"in_progress": "[yellow]in progress[/yellow]",
|
||
"complete": "[green]complete[/green]",
|
||
"error": "[red]error[/red]",
|
||
}
|
||
|
||
table = Table(box=box.SIMPLE, show_header=True, padding=(0, 2), expand=False)
|
||
table.add_column("Task", style="task_id", no_wrap=True)
|
||
table.add_column("Lead", no_wrap=True)
|
||
table.add_column("Status", no_wrap=True)
|
||
table.add_column("Brief", style="dim")
|
||
|
||
for t in tasks:
|
||
color = _agent_color(t.assigned_to)
|
||
status = STATUS_STYLE.get(t.status, t.status)
|
||
table.add_row(
|
||
t.task_id,
|
||
f"[{color}]{t.assigned_to.capitalize()}[/{color}]",
|
||
status,
|
||
_brief_preview(t.brief, max_len=60),
|
||
)
|
||
|
||
console.print(table)
|
||
console.print()
|
||
|
||
|
||
def print_debug_toggle(enabled: bool):
|
||
state = "[yellow]ON[/yellow]" if enabled else "[dim]OFF[/dim]"
|
||
console.print(f" [dim]Debug output:[/dim] {state}")
|
||
|
||
|
||
def print_audit_toggle(enabled: bool):
|
||
state = "[green]ON[/green]" if enabled else "[dim]OFF[/dim]"
|
||
console.print(f" [dim]Auto-audit (Vera):[/dim] {state}")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tool-call Live panel (transient, height-capped — one panel per tool loop)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
_TOOL_PANEL_VISIBLE = 8 # max call rows shown at once
|
||
|
||
class _ToolPanelState:
|
||
__slots__ = ("agent_name", "calls", "live")
|
||
def __init__(self, agent_name: str, live: "Live"):
|
||
self.agent_name = agent_name
|
||
self.calls: list[dict] = []
|
||
self.live = live
|
||
|
||
_active_tool_panel: "_ToolPanelState | None" = None
|
||
|
||
|
||
def _tool_result_summary(result_json: str) -> str:
|
||
"""Compact one-token summary of a tool result for the panel row."""
|
||
try:
|
||
obj = _json.loads(result_json)
|
||
if isinstance(obj, dict):
|
||
if "error" in obj:
|
||
return f"[red]error: {str(obj['error'])[:60]}[/red]"
|
||
if obj.get("ok"):
|
||
extra = obj.get("action") or obj.get("wrote") or obj.get("deleted") or obj.get("path", "")
|
||
return f"[green]ok[/green] [dim]{str(extra)[:50]}[/dim]" if extra else "[green]ok[/green]"
|
||
if "queued" in obj:
|
||
return f"[dim]queued → {obj.get('to', '')}[/dim]"
|
||
keys = ", ".join(list(obj.keys())[:4])
|
||
return f"[dim]{{{keys}{'…' if len(obj) > 4 else ''}}}[/dim]"
|
||
if isinstance(obj, list):
|
||
return f"[dim][{len(obj)} items][/dim]"
|
||
return f"[dim]{str(obj)[:60]}[/dim]"
|
||
except Exception:
|
||
return f"[dim]{result_json[:60]}[/dim]"
|
||
|
||
|
||
def _build_tool_panel_renderable(agent_name: str, calls: list[dict]) -> object:
|
||
color = _agent_color(agent_name)
|
||
title_text = _agent_display_title(agent_name)
|
||
|
||
if not calls:
|
||
return Panel(
|
||
Text(f" [{color} dim]{title_text} calling tools…[/{color} dim]", style="dim"),
|
||
title=f"[{color} dim]{title_text} — tools[/{color} dim]",
|
||
border_style="dim",
|
||
padding=(0, 1),
|
||
box=box.SIMPLE,
|
||
)
|
||
|
||
hidden = max(0, len(calls) - _TOOL_PANEL_VISIBLE)
|
||
visible = calls[-_TOOL_PANEL_VISIBLE:] if hidden else calls
|
||
|
||
rows = []
|
||
if hidden:
|
||
rows.append(Text(f" +{hidden} call{'s' if hidden != 1 else ''} earlier\n", style="dim italic"))
|
||
|
||
for c in visible:
|
||
summary = _tool_result_summary(c["result"])
|
||
rows.append(Text.from_markup(
|
||
f" [dim]⚙[/dim] [{color}]{c['tool']}[/{color}] [dim]→[/dim] {summary}"
|
||
))
|
||
|
||
content = Group(*rows) if len(rows) > 1 else rows[0]
|
||
return Panel(
|
||
content,
|
||
title=f"[{color} dim]{title_text} — tools[/{color} dim]",
|
||
border_style="dim",
|
||
padding=(0, 1),
|
||
box=box.SIMPLE,
|
||
)
|
||
|
||
|
||
def start_tool_panel(agent_name: str) -> None:
|
||
"""Open the transient tool-call Live panel at the start of a tool loop."""
|
||
global _active_tool_panel
|
||
live = Live(
|
||
_build_tool_panel_renderable(agent_name, []),
|
||
console=console,
|
||
transient=True,
|
||
vertical_overflow="crop",
|
||
refresh_per_second=15,
|
||
)
|
||
live.start()
|
||
_active_tool_panel = _ToolPanelState(agent_name=agent_name, live=live)
|
||
|
||
|
||
def record_tool_call(agent_name: str, tool_name: str, args_json: str, result_json: str) -> None:
|
||
"""Add a completed call to the active panel and refresh."""
|
||
global _active_tool_panel
|
||
if _active_tool_panel is None:
|
||
return
|
||
_active_tool_panel.calls.append({"tool": tool_name, "args": args_json, "result": result_json})
|
||
_active_tool_panel.live.update(
|
||
_build_tool_panel_renderable(agent_name, _active_tool_panel.calls)
|
||
)
|
||
|
||
|
||
def end_tool_panel(agent_name: str) -> None:
|
||
"""Close the transient panel and print a persistent one-liner summary."""
|
||
global _active_tool_panel
|
||
if _active_tool_panel is None:
|
||
return
|
||
calls = _active_tool_panel.calls
|
||
_active_tool_panel.live.stop()
|
||
_active_tool_panel = None
|
||
|
||
if not calls:
|
||
return # no tools called — nothing to summarise
|
||
|
||
color = _agent_color(agent_name)
|
||
title_text = _agent_display_title(agent_name)
|
||
n = len(calls)
|
||
names_str = ", ".join(c["tool"] for c in calls)
|
||
console.print(Text.assemble(
|
||
(" ▸ ", "dim"),
|
||
(title_text, f"dim {color}"),
|
||
(f" {n} tool call{'s' if n != 1 else ''} ", "dim"),
|
||
(f"[{names_str}]", "dim italic"),
|
||
))
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tool-call display (debug / fallback — used when no panel is active)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def print_tool_call(agent_name: str, tool_name: str, arguments_json: str, result_json: str):
|
||
"""
|
||
Verbose tool-call display for debug mode.
|
||
Normal operation uses the transient panel (start/record/end_tool_panel).
|
||
"""
|
||
color = _agent_color(agent_name)
|
||
console.print(
|
||
f" [dim]⚙[/dim] [{color}]{agent_name.capitalize()}[/{color}]"
|
||
f" [dim]›[/dim] [bold]{tool_name}[/bold]"
|
||
)
|
||
try:
|
||
args_pretty = _json.dumps(_json.loads(arguments_json), indent=2, ensure_ascii=False)
|
||
for line in args_pretty.splitlines():
|
||
console.print(f" [dim]{line}[/dim]")
|
||
except Exception:
|
||
console.print(f" [dim]{arguments_json}[/dim]")
|
||
try:
|
||
result_obj = _json.loads(result_json)
|
||
result_pretty = _json.dumps(result_obj, indent=2, ensure_ascii=False)
|
||
has_error = isinstance(result_obj, dict) and "error" in result_obj
|
||
style = "red" if has_error else "dim"
|
||
console.print(f" [dim]←[/dim]")
|
||
for line in result_pretty.splitlines():
|
||
console.print(f" [{style}]{line}[/{style}]")
|
||
except Exception:
|
||
console.print(f" [dim]← {result_json}[/dim]")
|