diff --git a/orchestration/__pycache__/config.cpython-311.pyc b/orchestration/__pycache__/config.cpython-311.pyc new file mode 100644 index 0000000..060aaa2 Binary files /dev/null and b/orchestration/__pycache__/config.cpython-311.pyc differ diff --git a/orchestration/__pycache__/orchestrator.cpython-311.pyc b/orchestration/__pycache__/orchestrator.cpython-311.pyc new file mode 100644 index 0000000..e26df96 Binary files /dev/null and b/orchestration/__pycache__/orchestrator.cpython-311.pyc differ diff --git a/orchestration/__pycache__/task_store.cpython-311.pyc b/orchestration/__pycache__/task_store.cpython-311.pyc new file mode 100644 index 0000000..81c95c7 Binary files /dev/null and b/orchestration/__pycache__/task_store.cpython-311.pyc differ diff --git a/orchestration/__pycache__/ui.cpython-311.pyc b/orchestration/__pycache__/ui.cpython-311.pyc new file mode 100644 index 0000000..5351aae Binary files /dev/null and b/orchestration/__pycache__/ui.cpython-311.pyc differ diff --git a/orchestration/orchestrator.py b/orchestration/orchestrator.py index f2db31b..3bb96dc 100644 --- a/orchestration/orchestrator.py +++ b/orchestration/orchestrator.py @@ -44,6 +44,13 @@ class ProviderClient: temperature: float, max_tokens: int) -> str: return self.call(system, messages, model, temperature, max_tokens) + async def call_async_streaming(self, system: str, messages: list[dict], model: str, + temperature: float, max_tokens: int): + """Async generator yielding ('thinking'|'content', text) chunks. + Default: emits the full response as a single 'content' chunk.""" + response = await self.call_async(system, messages, model, temperature, max_tokens) + yield ('content', response) + # --- OpenAI and any OpenAI-compatible endpoint --- # Covers AIPA_PROVIDER / {AGENT}_PROVIDER = openai or openai_compatible. @@ -71,6 +78,29 @@ class OpenAIClient(ProviderClient): self.call_async(system, messages, model, temperature, max_tokens) ) + async def call_async_streaming(self, system, messages, model, temperature, max_tokens): + """Stream via the OpenAI-compatible API, yielding thinking and content chunks. + Thinking tokens arrive in delta.reasoning_content (Qwen3, DeepSeek R1 style); + regular tokens arrive in delta.content.""" + full_messages = [{"role": "system", "content": system}] + messages + stream = await self.client.chat.completions.create( + model=model, + temperature=temperature, + max_tokens=max_tokens, + messages=full_messages, + extra_body=self.extra_body or None, + stream=True, + ) + async for chunk in stream: + if not chunk.choices: + continue + delta = chunk.choices[0].delta + thinking = getattr(delta, 'reasoning_content', None) + if thinking: + yield ('thinking', thinking) + if delta.content: + yield ('content', delta.content) + # --- Anthropic --- # Uncomment when ready to use. Install: pip install anthropic @@ -283,6 +313,8 @@ async def call_agent_async( agent: AgentState, user_message: str, dry_run: bool = False, + stream: bool = True, + stream_mode: str = "deliverable", ) -> str: if dry_run: return f"[DRY RUN] Would call {agent.name} ({config.agent_provider_name(agent.name)}) with: {user_message[:100]}..." @@ -290,19 +322,32 @@ async def call_agent_async( agent.add_user(user_message) try: - response = await agent.client.call_async( - system=agent.system_prompt, - messages=agent.messages_for_call(), - model=agent.model, - temperature=agent.temperature, - max_tokens=agent.max_tokens, - ) + if stream: + response = await ui.stream_agent_output( + agent.name, + agent.client.call_async_streaming( + system=agent.system_prompt, + messages=agent.messages_for_call(), + model=agent.model, + temperature=agent.temperature, + max_tokens=agent.max_tokens, + ), + mode=stream_mode, + ) + else: + response = await agent.client.call_async( + system=agent.system_prompt, + messages=agent.messages_for_call(), + model=agent.model, + temperature=agent.temperature, + max_tokens=agent.max_tokens, + ) except NotImplementedError as e: raise RuntimeError(str(e)) from e agent.add_assistant(response) - if config.DEBUG_PRINT_AGENT_OUTPUTS: + if config.DEBUG_PRINT_AGENT_OUTPUTS and not stream: ui.print_agent_panel(agent.name, response) return response @@ -357,7 +402,7 @@ async def dispatch_to_lead( if session_id: task_store.update_task(task, session_id) try: - output = await call_agent_async(lead, task.brief, dry_run=dry_run) + output = await call_agent_async(lead, task.brief, dry_run=dry_run, stream=False) task.output = output task.status = "complete" except Exception as e: @@ -424,7 +469,7 @@ async def run_auditor( f"{deliverable}" ) auditor.history = [] # Auditor is stateless — fresh call each time - return await call_agent_async(auditor, audit_request, dry_run=dry_run) + return await call_agent_async(auditor, audit_request, dry_run=dry_run, stream_mode="audit") # --------------------------------------------------------------------------- @@ -446,7 +491,7 @@ async def update_standing_brief( f"log entry at the top of Section 9. Archive anything older than " f"{config.SESSION_LOG_RETENTION} sessions." ) - return await call_agent_async(orchestrator, update_request, dry_run=dry_run) + return await call_agent_async(orchestrator, update_request, dry_run=dry_run, stream_mode="background") # --------------------------------------------------------------------------- @@ -502,7 +547,7 @@ class Session: f"Directive: {directive}" ) cos_response = await call_agent_async( - self.orchestrator, cos_prompt, dry_run=self.dry_run + self.orchestrator, cos_prompt, dry_run=self.dry_run, stream=True ) # ── Step 2: Parse any task briefs the orchestrator issued ─────────── @@ -532,10 +577,9 @@ class Session: f"Please synthesize these into a final deliverable for the Principal.\n\n" f"{lead_outputs_text}" ) - with ui.synthesis_progress(): - deliverable = await call_agent_async( - self.orchestrator, synthesis_prompt, dry_run=self.dry_run - ) + deliverable = await call_agent_async( + self.orchestrator, synthesis_prompt, dry_run=self.dry_run, stream=True + ) dispatched_leads = sorted({t.assigned_to for t in completed_tasks}) self.session_log.append( @@ -549,10 +593,9 @@ class Session: audit_memo = "" if self.auto_audit: try: - with ui.audit_progress(): - audit_memo = await run_auditor( - self.auditor, deliverable, task_id, dry_run=self.dry_run - ) + audit_memo = await run_auditor( + self.auditor, deliverable, task_id, dry_run=self.dry_run + ) self.session_log.append(f"Audit complete for {task_id}.") except RuntimeError as e: ui.print_warning(f"Audit skipped — {e}") @@ -561,10 +604,9 @@ class Session: async def close(self): summary = "\n".join(self.session_log) - with ui.brief_update_progress(): - updated_brief = await update_standing_brief( - self.orchestrator, summary, dry_run=self.dry_run - ) + updated_brief = await update_standing_brief( + self.orchestrator, summary, dry_run=self.dry_run + ) if not self.dry_run: save_standing_brief(updated_brief) ui.print_session_footer(self.directive_count) @@ -645,17 +687,26 @@ async def cmd_history(session: Session, args: str) -> bool: return True +@command(["/thinking"], "Show the thinking output from the last response") +async def cmd_thinking(session: Session, args: str) -> bool: + last = ui.get_last_thinking() + if not last: + ui.print_system("No thinking output recorded — thinking mode may not be active.") + else: + ui.print_thinking_expansion(last) + return True + + @command(["/audit"], "Re-run Vera's audit on the last deliverable") async def cmd_audit(session: Session, args: str) -> bool: if not session.last_deliverable: ui.print_warning("No deliverable yet — issue a directive first.") return True ui.print_system("Running audit on last deliverable…") - with ui.audit_progress(): - memo = await run_auditor( - session.auditor, session.last_deliverable, - session.last_task_id, dry_run=session.dry_run, - ) + memo = await run_auditor( + session.auditor, session.last_deliverable, + session.last_task_id, dry_run=session.dry_run, + ) ui.print_audit_memo(memo, task_id=session.last_task_id) return True @@ -731,7 +782,15 @@ async def direct_agent_session(agent_name: str, dry_run: bool = False): if not user_input or user_input.lower() in ("/quit", "/exit", "/q"): break - response = await call_agent_async(agent, user_input, dry_run=dry_run) + if user_input.lower() == "/thinking": + last = ui.get_last_thinking() + if not last: + ui.print_system("No thinking output recorded — thinking mode may not be active.") + else: + ui.print_thinking_expansion(last) + continue + + response = await call_agent_async(agent, user_input, dry_run=dry_run, stream=True) ui.print_agent_panel(agent_name, response) ui.print_session_footer(0) diff --git a/orchestration/ui.py b/orchestration/ui.py index cdcbba7..5bf7de4 100644 --- a/orchestration/ui.py +++ b/orchestration/ui.py @@ -23,8 +23,9 @@ from datetime import datetime import config as _config -from rich.console import Console +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 ( @@ -75,6 +76,10 @@ def _build_theme() -> Theme: console = Console(theme=_build_theme(), highlight=False) +_last_thinking: str = "" # Stored from most recent streamed response +_last_stream_rendered: bool = False # True when stream_agent_output rendered the final panel + + # --------------------------------------------------------------------------- # Agent visual config # --------------------------------------------------------------------------- @@ -268,7 +273,13 @@ 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 "" @@ -286,15 +297,51 @@ def print_agent_panel(agent_name: str, content: str, task_id: str = ""): def print_deliverable(content: str, task_id: str = ""): - """Orchestrator's synthesized deliverable.""" - print_agent_panel(_config.ORCHESTRATOR_AGENT, content, task_id) + """ + 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) @@ -390,6 +437,267 @@ def brief_update_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_renderable( + thinking_text: str, + active: bool, + agent_name: str, + mode: str, + window: int = 300, +) -> object: + """ + Build the thinking panel (active) or collapsed summary line (done). + + Active: dim panel with a rolling window of the last `window` chars, + plus an overflow indicator when earlier content has scrolled off. + Collapsed: single dim line — ' ▸ Thinking (N chars, L lines)'. + """ + if not thinking_text: + return None + + if active: + overflow = len(thinking_text) - window + if overflow > 0: + preview = thinking_text[-window:] + header = Text(f" +{overflow} chars earlier\n", style="dim italic") + body = Text(preview, style="dim") + content = Group(header, body) + else: + content = Text(thinking_text, style="dim") + + return Panel( + content, + title=_thinking_panel_title(agent_name, mode), + border_style="dim", + padding=(0, 1), + box=box.SIMPLE, + ) + else: + lines = thinking_text.count("\n") + 1 + return Text( + f" ▸ Thinking [{len(thinking_text):,} chars · {lines} lines]", + style="dim", + ) + + +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. + + mode values + ----------- + "deliverable" Normal agent response — thinking panel + full styled panel. + The Live display is NOT transient: the final Markdown panel + is rendered inside the Live block, so there is no erase/ + reprint flash. Sets _last_stream_rendered = True so that + the caller's print_deliverable / print_agent_panel becomes + a no-op. + + "audit" Like "deliverable" but uses Vera's yellow identity cues + during the thinking phase. Sets _last_stream_rendered so + print_audit_memo is a no-op (the verdict-border panel is + rendered here instead). + + "background" Housekeeping call (e.g. standing-brief update). A compact + transient progress line is shown; no full panel is printed. + _last_stream_rendered is set so the caller's print_* is a + no-op. + + In all modes the full content string is returned and _last_thinking is + updated so /thinking works. + + Thinking tokens stream inside a dim panel. When the first content token + arrives the panel collapses to a single dim summary line and content begins + streaming in the agent's normal styled panel. + + The full thinking text is stored in _last_thinking and can be retrieved + with get_last_thinking() (exposed to the /thinking command). + + Returns the full content string. + """ + global _last_thinking, _last_stream_rendered + + color = _agent_color(agent_name) + title_text = _agent_display_title(agent_name) + + # ── Background mode: compact transient progress line ──────────────────── + 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) + + _last_thinking = "".join(thinking_parts) + _last_stream_rendered = True + return "".join(content_parts) + + # ── Deliverable / audit modes ──────────────────────────────────────────── + thinking_parts: list[str] = [] + content_parts: list[str] = [] + phase = "start" # "start" | "thinking" | "content" + + def _build_final_panel(content_text: str) -> Panel: + """Render the agent's finished output as a polished Markdown panel.""" + if mode == "audit": + # Re-derive verdict border from completed content + 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, + ) + else: + return Panel( + Markdown(content_text), + title=f"[{color}] {title_text} [/{color}]", + border_style=color, + padding=(1, 2), + box=box.ROUNDED, + ) + + def _renderable(final: bool = False): + """ + Build the live renderable for the current stream state. + + final=True: replace plain Text content with Markdown and apply the + verdict-aware border so the Live display ends in its final polished + form. This avoids a transient erase + reprint flash. + """ + thinking_text = "".join(thinking_parts) + content_text = "".join(content_parts) + items: list = [] + + # ── Agent identity line (shown while only thinking, not yet content) ── + if phase in ("start", "thinking") and not content_text: + items.append(Text( + f" {title_text}", + style=f"dim {color}", + )) + + # ── Thinking panel / collapsed summary ────────────────────────────── + thinking_active = (phase == "thinking") + think_renderable = _build_thinking_renderable( + thinking_text, thinking_active, agent_name, mode + ) + if think_renderable is not None: + items.append(think_renderable) + + # ── Content panel ──────────────────────────────────────────────────── + if content_text or phase == "content": + if final and content_text: + items.append(_build_final_panel(content_text)) + else: + items.append(Panel( + Text(content_text), + title=f"[{color}] {title_text} [/{color}]", + border_style=color, + padding=(1, 2), + box=box.ROUNDED, + )) + + if not items: + # Nothing yet — show a faint connecting indicator + return Text( + f" [{_agent_color(agent_name)} dim]{title_text} …[/{_agent_color(agent_name)} dim]", + style="dim", + ) + return Group(*items) if len(items) > 1 else items[0] + + console.print() + # transient=False: the Live display persists — we render the final panel + # inside it before stopping, so there is no erase/reprint cycle. + with Live( + _renderable(), + console=console, + refresh_per_second=15, + transient=False, + vertical_overflow="visible", + ) as live: + async for chunk_type, text in stream_gen: + if chunk_type == "thinking": + if phase == "start": + phase = "thinking" + thinking_parts.append(text) + elif chunk_type == "content": + if phase in ("start", "thinking"): + phase = "content" + content_parts.append(text) + live.update(_renderable()) + + # Final update: replace plain Text with formatted Markdown panel + # (and apply verdict border for audit mode) before the Live context + # exits and the display freezes in place. + live.update(_renderable(final=True)) + + _last_thinking = "".join(thinking_parts) + _last_stream_rendered = True + return "".join(content_parts) + + +def get_last_thinking() -> str: + """Return the thinking text captured from the most recent streamed response.""" + return _last_thinking + + +def print_thinking_expansion(thinking_text: str): + """Render the stored thinking output in a dim panel.""" + panel = Panel( + Text(thinking_text, style="dim"), + title="[dim] Thinking — last response [/dim]", + border_style="dim", + padding=(1, 2), + box=box.ROUNDED, + ) + console.print() + console.print(panel) + + # --------------------------------------------------------------------------- # Dry-run callout # ---------------------------------------------------------------------------