Streaming Progress Compositor

Fold typed StreamEvents into a bounded multi-line status view

The progress compositor folds an agent’s typed StreamEvent stream into a bounded, multi-line status view you can render in any transport — bots today, TUI or web tomorrow.

The compositor is pure stdlib — zero third-party dependencies, deterministic, and side-effect free. Every call returns a new list; the input is never mutated. The same three exports power the bot progress_style="feed" view.

Quick Start

1. Fold events into lines

   from praisonaiagents.streaming import merge_progress_line, render_progress
   
   lines = []
   for event in agent_events:          # any typed StreamEvent stream
       lines = merge_progress_line(lines, event)
   
   print(render_progress(lines))

2. Wire it into your own callback

   from praisonaiagents import Agent
   from praisonaiagents.streaming import (
       ProgressLine, merge_progress_line, render_progress,
   )
   
   lines: list[ProgressLine] = []
   
   def my_callback(event) -> None:
       global lines
       lines = merge_progress_line(lines, event)
       print("\033[2J\033[H" + render_progress(lines, max_lines=6))
   
   agent = Agent(name="research-assistant", instructions="Research and summarise.")
   agent.stream_emitter.add_callback(my_callback)
   agent.start("Research quantum computing")

   Any typed StreamEvent stream — even one you produce yourself — folds into the same bounded view.

How It Works

A start event and its matching finish event share one correlation id, so merge_progress_line updates the existing line instead of appending a duplicate.

Each event type maps to a line kind and state:

Configuration Options

render_progress bounds the output for edit-in-place delivery.

Each ProgressLine carries the state for a single step.

State machine rules

Terminal states are sticky, so the feed never lies about an outcome.

• Error takes precedence — once a line is error (✗) it stays error, even if a later done event arrives for the same id.
• Done is not downgraded — a late running event never resets a completed done (✓) line back to running.
• Non-progress events are no-ops — events outside the mapping table return the input list unchanged, so unrelated stream traffic never adds noise.

Common Patterns

Build a live TUI panel straight from the agent’s event emitter.

from praisonaiagents import Agent
from praisonaiagents.streaming import merge_progress_line, render_progress

agent = Agent(name="assistant", instructions="Use tools and summarise.")
lines = []

def render_panel(event):
    global lines
    lines = merge_progress_line(lines, event)
    print("\033[2J\033[H" + render_progress(lines, max_lines=10))

agent.stream_emitter.add_callback(render_panel)
agent.start("Compare three databases")

Feed the compositor from your own event source — plan steps, approvals, or command output all fold in the same way, since merge_progress_line only cares about the event’s type and correlation id.

Override the glyphs by wrapping render_progress with a renderer that walks each ProgressLine.state.

from praisonaiagents.streaming import ProgressLine

GLYPHS = {"running": "•", "done": "[OK]", "error": "[X]"}

def render_ascii(lines: list[ProgressLine]) -> str:
    return "\n".join(f"{GLYPHS.get(l.state, '•')} {l.text}" for l in lines[-8:])

Best Practices

Related