feat(replay): watch() / replay_app(): a Run button that draws each run live, and embedding

examples/replay/README.md

@@ -0,0 +1,16 @@
+# PraisonAI Replay: watch runs live
+
+| Example | What it shows |
+|---|---|
+| `watch_flow.py` | Click **Run** and watch each step light up on the map; **Play** goes through it again. No API key. |
+| `watch_team.py` | The same for two real agents (needs `OPENAI_API_KEY`). |
+| `embed_fastapi.py` | Your own app starts a run and shows its progress in an `<iframe>` (`app.mount("/replay", replay_app())`). |
+
+The same from the command line, for a YAML file:
+
+```bash
+praisonai replay watch agents.yaml
+```
+
+Install: `pip install "praisonai[ui]"`. Keep the page on localhost or behind your app's
+login: anyone who can reach it can read the runs, and with a Run button, start one.

examples/replay/embed_fastapi.py

@@ -0,0 +1,60 @@
+"""Show agent progress inside your own web app.
+
+Your app starts the run and records it; the Replay view, mounted at /replay,
+draws it live in an <iframe>. No dashboard, no second server.
+
+    pip install "praisonai[ui]" fastapi uvicorn
+    uvicorn embed_fastapi:app --port 8000      # open http://127.0.0.1:8000/
+"""
+
+import threading
+import time
+import uuid
+
+from fastapi import FastAPI
+from fastapi.responses import HTMLResponse
+from praisonaiagents import AgentFlow
+from praisonaiagents.workflows import StepResult
+from praisonai.replay import record, replay_app
+
+
+def research(ctx):
+    time.sleep(1.2)
+    return StepResult(output="notes")
+
+
+def write(ctx):
+    time.sleep(1.2)
+    return StepResult(output="article")
+
+
+flow = AgentFlow(name="Article", steps=[research, write])
+
+app = FastAPI()
+app.mount("/replay", replay_app())  # read-only view: /replay/?run=<id> shows one run
+
+
+@app.post("/generate")
+def generate():
+    run_id = f"run-{uuid.uuid4().hex[:12]}"
+
+    def job():
+        with record(run_id, live=True):  # live: the view follows it as it is written
+            flow.start("solar sails")
+
+    threading.Thread(target=job, daemon=True).start()
+    return {"run_id": run_id}
+
+
+@app.get("/", response_class=HTMLResponse)
+def home():
+    return """<!doctype html><title>My app</title>
+<h1>My app</h1>
+<button id="go">Generate article</button>
+<iframe id="progress" title="Progress" style="width:100%;height:640px;border:1px solid #ccc"></iframe>
+<script>
+  go.onclick = async () => {
+    const { run_id } = await (await fetch('/generate', { method: 'POST' })).json();
+    progress.src = '/replay/?run=' + run_id;
+  };
+</script>"""

examples/replay/watch_flow.py

@@ -0,0 +1,40 @@
+"""Watch a flow run, step by step: click Run and each step lights up on the map.
+
+No API key needed (the steps are plain Python that take a moment each).
+
+    pip install "praisonai[ui]"
+    python watch_flow.py          # opens http://127.0.0.1:8090/
+
+Map: dashed = not reached, blue = running now, green = done.
+After a run, press Play to go through it again one step at a time.
+"""
+
+import time
+
+from praisonaiagents import AgentFlow
+from praisonaiagents.workflows import StepResult
+from praisonai.replay import watch
+
+
+def research(ctx):
+    time.sleep(1.5)
+    return StepResult(output=f"3 sources about {ctx.input}")
+
+
+def outline(ctx):
+    time.sleep(1.5)
+    return StepResult(output="intro, body, close")
+
+
+def draft(ctx):
+    time.sleep(1.5)
+    return StepResult(output="first draft")
+
+
+def review(ctx):
+    time.sleep(1.5)
+    return StepResult(output="approved")
+
+
+flow = AgentFlow(name="Write an article", steps=[research, outline, draft, review])
+watch(flow, input="solar sails")

examples/replay/watch_team.py

@@ -0,0 +1,18 @@
+"""Watch two agents work together, live (calls the LLM: needs OPENAI_API_KEY).
+
+    python watch_team.py          # opens http://127.0.0.1:8090/, press Run
+"""
+
+from praisonaiagents import Agent, AgentTeam, Task
+from praisonai.replay import watch
+
+researcher = Agent(name="Researcher", instructions="Give 3 short facts.", llm="gpt-4o-mini")
+writer = Agent(name="Writer", instructions="Write one short paragraph from the facts.", llm="gpt-4o-mini")
+team = AgentTeam(
+    agents=[researcher, writer],
+    tasks=[
+        Task(name="research", description="3 facts about solar sails", agent=researcher, expected_output="3 facts"),
+        Task(name="write", description="One paragraph using the facts", agent=writer, expected_output="a paragraph"),
+    ],
+)
+watch(team)

src/praisonai/praisonai/cli/commands/replay.py

@@ -6,6 +6,7 @@
 - replay context <session_id>: Interactive replay of a trace
 - replay flow <run|yaml> --format ascii|json|mermaid: Agent flow / static graph
 - replay view [run|yaml] [--live]: Open PraisonAI Replay (aiui) on a run
+- replay watch [yaml]: A page with a Run button that draws each run live
 - replay export <run> --format html|json|mermaid|otel: Export a run
 - replay rerun <run> --from NODE: Re-run from a node, reusing earlier outputs
 """
@@ -1100,6 +1101,35 @@ def _load_raw_events(output, run: str):
     return events
 
 
+@app.command("watch")
+def replay_watch(
+    yaml_file: Optional[str] = typer.Argument(None, help="agents.yaml to run from the page (default: just view recorded runs)"),
+    port: int = typer.Option(8090, "--port", "-p", help="Port for the page"),
+    host: str = typer.Option("127.0.0.1", "--host", help="Host to bind to"),
+    no_browser: bool = typer.Option(False, "--no-browser", help="Do not open the browser"),
+):
+    """Open a page with a Run button: each run is recorded and drawn live, step by step.
+
+    Examples:
+        praisonai replay watch agents.yaml
+        praisonai replay watch               # recorded runs only
+    """
+    from praisonai.replay.live import watch
+    from praisonai.replay.yaml_run import is_yaml_path
+
+    if not _aiui_installed():
+        typer.echo(f"ERROR: PraisonAI Replay needs PraisonAI UI (aiui). Install with:\n  {_AIUI_INSTALL_HINT}", err=True)
+        raise typer.Exit(1)
+    if yaml_file and not is_yaml_path(yaml_file):
+        typer.echo(f"ERROR: not a YAML file: {yaml_file}", err=True)
+        raise typer.Exit(2)
+    try:
+        watch(yaml_file, host=host, port=port, open_browser=not no_browser)
+    except ImportError as e:
+        typer.echo(f"ERROR: {e}", err=True)
+        raise typer.Exit(1)
+
+
 @app.command("export")
 def replay_export(
     run: str = typer.Argument(..., help="Run id or path to trace file"),

src/praisonai/praisonai/replay/__init__.py

@@ -29,11 +29,18 @@
     with record(live=True) as rec:
         agent.start("Say hello")
     html = export.to_html(load_events(rec.path))
+
+    # Watch it live: a page with a Run button (or mount it in your own app)
+    from praisonai.replay import watch, replay_app
+    watch(flow)                                  # opens the browser
+    app.mount("/replay", replay_app(flow))       # FastAPI / Starlette
 """
 
 __all__ = [
     "record",
     "Recording",
+    "watch",
+    "replay_app",
     "TraceStepCache",
     "load_events",
     "ContextTraceWriter",
@@ -57,6 +64,10 @@ def __getattr__(name: str):
         from . import recorder
         return getattr(recorder, name)
 
+    if name in ("watch", "replay_app"):
+        from . import live
+        return getattr(live, name)
+
     if name == "TraceStepCache":
         from .rerun import TraceStepCache
         return TraceStepCache

src/praisonai/praisonai/replay/live.py

@@ -0,0 +1,138 @@
+"""Watch an agent, team or flow run live: a page with a Run button (PraisonAI Replay).
+
+    from praisonai.replay import watch
+    watch(flow)                     # opens the browser; each Run is recorded and drawn live
+
+    from praisonai.replay import replay_app
+    app.mount("/replay", replay_app(flow))   # inside your own FastAPI/Starlette app
+
+The page is aiui's Replay view without the dashboard around it
+(``praisonaiui.features.replay.create_replay_app``). Requires ``pip install "praisonai[ui]"``.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import inspect
+import threading
+import webbrowser
+from pathlib import Path
+from typing import Any, Callable, Optional
+
+_LOOPBACK_HOSTS = {"127.0.0.1", "localhost", "::1"}
+_INSTALL_HINT = 'PraisonAI Replay needs PraisonAI UI: pip install -U "praisonai[ui]"'
+
+
+def _runner(target: Any) -> Optional[Callable[[str, str], None]]:
+    """``run(input, run_id)`` that records *target* under ``run_id``, live."""
+    if target is None:
+        return None
+    from .recorder import record
+    from .yaml_run import is_yaml_path, run_yaml_recorded
+
+    if isinstance(target, (str, Path)):
+        if not is_yaml_path(str(target)):
+            raise ValueError(f"not a YAML file: {target}")
+        path = str(target)
+
+        def run_yaml(_text: str, run_id: str) -> None:
+            run_yaml_recorded(path, run_id=run_id, live=True)
+
+        return run_yaml
+
+    start = getattr(target, "start", None)
+    call = start if callable(start) else target
+    if not callable(call):
+        raise TypeError("target must be an Agent, AgentTeam, AgentFlow, a YAML path or a callable(input)")
+
+    # An AgentTeam skips tasks it already completed and appends each input to
+    # every task's context; each Run starts them afresh with the original context.
+    reset = getattr(target, "_reset_task_state_for_item", None)
+    tasks = getattr(target, "tasks", None)
+    tasks = list(tasks.values()) if isinstance(tasks, dict) and callable(reset) else []
+    contexts = [(t, list(t.context) if isinstance(getattr(t, "context", None), list) else None) for t in tasks]
+
+    def run(text: str, run_id: str) -> None:
+        if callable(reset):
+            reset()
+        for task, context in contexts:
+            task.context = list(context) if context is not None else None
+        with record(run_id, live=True):
+            out = call(text)
+            if inspect.isawaitable(out):
+                asyncio.run(out)
+
+    return run
+
+
+def _title(target: Any) -> str:
+    if isinstance(target, (str, Path)):
+        return f"PraisonAI Replay: {Path(target).name}"
+    name = getattr(target, "name", None) or getattr(target, "__name__", None)
+    return f"PraisonAI Replay: {name}" if isinstance(name, str) and name else "PraisonAI Replay"
+
+
+def replay_app(target: Any = None, *, input: str = "", placeholder: str = "", title: Optional[str] = None):
+    """The Replay page as an ASGI app; with *target*, it has a Run button.
+
+    Args:
+        target: An Agent, AgentTeam or AgentFlow (``.start(input)`` is called), a
+            YAML file (run as written; no input box), or any ``callable(input)``.
+            ``None`` gives a read-only view of recorded runs.
+        input: Text the input box starts with.
+        placeholder: Placeholder for the input box.
+        title: Page title (default: from the target's name).
+
+    Each Run is recorded with :func:`praisonai.replay.record` under a new run id
+    and the page follows it live. One run at a time. The same *target* object is
+    reused: a team's tasks start afresh each run, an Agent keeps its chat history.
+
+    Open ``<mount>/`` for the run list, or ``<mount>/?run=<run_id>`` for one run
+    (handy in an ``<iframe>``).
+    """
+    try:
+        from praisonaiui.features.replay import create_replay_app
+    except ImportError as e:  # aiui missing or older than the embeddable app
+        raise ImportError(_INSTALL_HINT) from e
+    is_yaml = isinstance(target, (str, Path))
+    return create_replay_app(
+        _runner(target),
+        title=title or _title(target),
+        input=None if is_yaml else input,
+        placeholder=placeholder,
+    )
+
+
+def watch(
+    target: Any = None,
+    *,
+    input: str = "",
+    placeholder: str = "",
+    title: Optional[str] = None,
+    host: str = "127.0.0.1",
+    port: int = 8090,
+    open_browser: bool = True,
+) -> None:
+    """Serve :func:`replay_app` and open it in the browser (blocks until Ctrl+C).
+
+    Example::
+
+        from praisonai.replay import watch
+        watch(flow, input="Write about solar sails")
+    """
+    import uvicorn
+
+    app = replay_app(target, input=input, placeholder=placeholder, title=title)
+    if host not in _LOOPBACK_HOSTS:
+        print(
+            f"WARNING: serving on {host} lets anyone who can reach it read your runs"
+            + (" and start new ones." if target is not None else ".")
+        )
+    browser_host = "127.0.0.1" if host in ("0.0.0.0", "::") else host
+    if ":" in browser_host and not browser_host.startswith("["):
+        browser_host = f"[{browser_host}]"
+    url = f"http://{browser_host}:{port}/"
+    print(f"PraisonAI Replay: {url}  (Ctrl+C to stop)")
+    if open_browser:
+        threading.Timer(1.0, webbrowser.open, [url]).start()
+    uvicorn.run(app, host=host, port=port, log_level="warning")

src/praisonai/tests/unit/replay/test_replay_cli.py

@@ -345,3 +345,35 @@ def test_team_yaml_static_graph_uses_the_run_ids(tmp_path):
     ids = {n["id"] for n in graph["nodes"]}
     assert {"agent:Researcher", "agent:Writer", "task:0", "task:1"} <= ids
     assert not any(i.startswith("step:") for i in ids)
+
+
+# ---------------------------------------------------------------------------
+# watch: a page with a Run button (the server is never started here)
+# ---------------------------------------------------------------------------
+
+def test_watch_needs_aiui(monkeypatch):
+    from praisonai.cli.commands import replay as replay_cli
+
+    monkeypatch.setattr(replay_cli, "_aiui_installed", lambda: False)
+    result = _invoke("watch", "agents.yaml")
+    assert result.exit_code == 1 and "praisonai[ui]" in result.output
+
+
+def test_watch_rejects_a_non_yaml_file(monkeypatch):
+    from praisonai.cli.commands import replay as replay_cli
+
+    monkeypatch.setattr(replay_cli, "_aiui_installed", lambda: True)
+    result = _invoke("watch", "notes.txt")
+    assert result.exit_code == 2 and "not a YAML file" in result.output
+
+
+def test_watch_serves_the_yaml_with_a_run_button(monkeypatch):
+    from praisonai.cli.commands import replay as replay_cli
+    from praisonai.replay import live
+
+    calls = []
+    monkeypatch.setattr(replay_cli, "_aiui_installed", lambda: True)
+    monkeypatch.setattr(live, "watch", lambda target, **kw: calls.append((target, kw)))
+    result = _invoke("watch", "agents.yaml", "--port", "8123", "--no-browser")
+    assert result.exit_code == 0, result.output
+    assert calls == [("agents.yaml", {"host": "127.0.0.1", "port": 8123, "open_browser": False})]