1113 lines
39 KiB
Python
1113 lines
39 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)
|
||
|
||
_last_stream_rendered: bool = False # True when stream_agent_output rendered the final panel
|
||
|
||
# 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
|
||
|
||
|
||
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]
|
||
|
||
# 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)
|
||
|
||
|
||
@contextmanager
|
||
def lead_dispatch_progress(tasks: list):
|
||
"""
|
||
Per-task live progress display while leads are working.
|
||
Yields (progress, rows) where rows is keyed by task_id.
|
||
"""
|
||
def _row(task_id: str, lead_name: str, status: str) -> str:
|
||
color = _agent_color(lead_name)
|
||
return (
|
||
f"[task_id]{task_id}[/task_id] "
|
||
f"[{color}]{lead_name.capitalize()}[/{color}] "
|
||
f"{status}"
|
||
)
|
||
|
||
with Progress(
|
||
SpinnerColumn(style="dim"),
|
||
TextColumn("{task.description}"),
|
||
TimeElapsedColumn(),
|
||
console=console,
|
||
transient=True,
|
||
) as progress:
|
||
rows = {
|
||
t.task_id: progress.add_task(
|
||
_row(t.task_id, t.assigned_to, "[dim]queued[/dim]"),
|
||
total=1,
|
||
)
|
||
for t in tasks
|
||
}
|
||
yield progress, rows
|
||
|
||
|
||
def mark_task_running(progress, rows: dict, task_id: str, lead_name: str):
|
||
"""Update a task row to show it is actively being worked."""
|
||
if task_id not in rows:
|
||
return
|
||
color = _agent_color(lead_name)
|
||
progress.update(
|
||
rows[task_id],
|
||
description=(
|
||
f"[task_id]{task_id}[/task_id] "
|
||
f"[{color}]{lead_name.capitalize()}[/{color}] "
|
||
f"[yellow]working…[/yellow]"
|
||
),
|
||
)
|
||
|
||
|
||
def mark_task_done(progress, rows: dict, task_id: str, lead_name: str, error: bool = False):
|
||
"""Update a task row to show it is finished (complete or error)."""
|
||
if task_id not in rows:
|
||
return
|
||
color = _agent_color(lead_name)
|
||
status = "[red]error[/red]" if error else "[green]done[/green]"
|
||
progress.update(
|
||
rows[task_id],
|
||
completed=1,
|
||
description=(
|
||
f"[task_id]{task_id}[/task_id] "
|
||
f"[{color}]{lead_name.capitalize()}[/{color}] "
|
||
f"{status}"
|
||
),
|
||
)
|
||
|
||
|
||
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 _last_stream_rendered
|
||
if _last_stream_rendered:
|
||
_last_stream_rendered = 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 _last_stream_rendered
|
||
if _last_stream_rendered:
|
||
_last_stream_rendered = 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 _last_stream_rendered
|
||
if _last_stream_rendered:
|
||
_last_stream_rendered = 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)
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 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
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 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 _last_stream_rendered
|
||
|
||
color = _agent_color(agent_name)
|
||
title_text = _agent_display_title(agent_name)
|
||
|
||
# ── 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))
|
||
_last_stream_rendered = 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))
|
||
_last_stream_rendered = 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))
|
||
_last_stream_rendered = 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 _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 display
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def print_tool_call(agent_name: str, tool_name: str, arguments_json: str, result_json: str):
|
||
"""
|
||
Render a single tool invocation inline during an agent's tool-calling loop.
|
||
|
||
Layout:
|
||
⚙ AgentName › tool_name
|
||
arguments (pretty-printed JSON, dim)
|
||
← result (pretty-printed JSON; red on parse error)
|
||
"""
|
||
color = _agent_color(agent_name)
|
||
|
||
# Header line
|
||
console.print(
|
||
f" [dim]⚙[/dim] [{color}]{agent_name.capitalize()}[/{color}]"
|
||
f" [dim]›[/dim] [bold]{tool_name}[/bold]"
|
||
)
|
||
|
||
# Arguments
|
||
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]")
|
||
|
||
# Result
|
||
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]")
|