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

- praisonai.replay.watch(target): serves a page with a Run button; each click records the Agent/AgentTeam/AgentFlow/YAML/callable under a new run id and the page draws it live (map: not reached / running / done). A team's tasks start afresh each run (AgentTeam otherwise skips completed tasks). - praisonai.replay.replay_app(target=None): the same page as an ASGI app to mount in your own FastAPI/Starlette app; /replay/?run=<id> shows one run (iframe). - CLI: praisonai replay watch [agents.yaml] ("show" was taken). - examples/replay: watch_flow.py (no key), watch_team.py, embed_fastapi.py. Needs aiui with create_replay_app (MervinPraison/PraisonAIUI#317); older aiui gets a clear "pip install -U praisonai[ui]" error and the tests skip. Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
2026-11-07 04:35:58 -10:00 · 2026-10-26 15:47:17 +00:00
1 parent 930d6d9837
commit faba52b940
@@ -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.
@@ -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>"""
@@ -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")
@@ -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)
@@ -6,6 +6,7 @@ Provides context replay functionality:
 - 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"),
@@ -29,11 +29,18 @@ Usage:
    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
@@ -0,0 +1,125 @@
+"""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 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; each Run starts them afresh.
+    reset = getattr(target, "_reset_task_state_for_item", None)
+
+    def run(text: str, run_id: str) -> None:
+        if callable(reset):
+            reset()
+        with record(run_id, live=True):
+            call(text)
+
+    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 ".")
+        )
+    url = f"http://{'127.0.0.1' if host in ('0.0.0.0', '::') else 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")
@@ -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})]
@@ -0,0 +1,135 @@
+"""praisonai.replay.replay_app / watch: a page with a Run button that records each run live."""
+
+import json
+import sys
+import threading
+
+import pytest
+
+
+def _flow(seen=None):
+    from praisonaiagents import AgentFlow
+    from praisonaiagents.workflows import StepResult
+
+    def research(ctx):
+        if seen is not None:
+            seen.append(ctx.input)
+        return StepResult(output="notes")
+
+    def write(ctx):
+        return StepResult(output="article")
+
+    return AgentFlow(name="Article", steps=[research, write])
+
+
+def _events(path):
+    return [json.loads(line) for line in path.read_text().splitlines() if line.strip()]
+
+
+def test_runner_records_a_flow_under_the_given_run_id(traces_dir):
+    from praisonai.replay.live import _runner
+
+    seen = []
+    _runner(_flow(seen))("solar sails", "run-live00000001")
+    path = traces_dir / "run-live00000001.jsonl"
+    types = [e["event_type"] for e in _events(path)]
+    assert seen == ["solar sails"]
+    assert types[0] == "session_start" and types[-1] == "session_end"
+    assert types.count("step_end") == 2
+
+
+def test_runner_resets_a_team_before_each_run(traces_dir):
+    from praisonai.replay.live import _runner
+
+    calls = []
+
+    class Team:
+        def _reset_task_state_for_item(self):
+            calls.append("reset")
+
+        def start(self, text):
+            calls.append(("start", text))
+
+    run = _runner(Team())
+    run("a", "run-team-a")
+    run("b", "run-team-b")
+    assert calls == ["reset", ("start", "a"), "reset", ("start", "b")]
+
+
+def test_runner_accepts_a_plain_callable(traces_dir):
+    from praisonai.replay.live import _runner
+
+    got = []
+    _runner(lambda text: got.append(text))("hi", "run-fn")
+    assert got == ["hi"] and (traces_dir / "run-fn.jsonl").is_file()
+
+
+def test_runner_runs_a_yaml_file_as_written(monkeypatch, tmp_path):
+    from praisonai.replay import live, yaml_run
+
+    calls = []
+    monkeypatch.setattr(yaml_run, "run_yaml_recorded", lambda path, **kw: calls.append((path, kw)))
+    yml = tmp_path / "agents.yaml"
+    yml.write_text("roles: {}\n")
+    live._runner(str(yml))("ignored", "run-yaml")
+    assert calls == [(str(yml), {"run_id": "run-yaml", "live": True})]
+
+
+def test_runner_rejects_bad_targets():
+    from praisonai.replay.live import _runner
+
+    assert _runner(None) is None
+    with pytest.raises(ValueError, match="not a YAML file"):
+        _runner("notes.txt")
+    with pytest.raises(TypeError, match="Agent, AgentTeam, AgentFlow"):
+        _runner(42)
+
+
+def test_replay_app_explains_how_to_install_aiui(monkeypatch):
+    from praisonai.replay import replay_app
+
+    monkeypatch.setitem(sys.modules, "praisonaiui.features.replay", None)
+    with pytest.raises(ImportError, match=r'pip install -U "praisonai\[ui\]"'):
+        replay_app()
+
+
+def _aiui_embed():
+    mod = pytest.importorskip("praisonaiui.features.replay")
+    if not hasattr(mod, "create_replay_app"):
+        pytest.skip("installed aiui predates create_replay_app")
+    return mod
+
+
+def test_run_button_records_and_serves_the_run_end_to_end(traces_dir):
+    """Mounted like app.mount('/replay', replay_app(flow)): Run, then read the run back."""
+    _aiui_embed()
+    from starlette.applications import Starlette
+    from starlette.routing import Mount
+    from starlette.testclient import TestClient
+
+    from praisonai.replay import replay_app
+
+    traces_dir.mkdir(parents=True, exist_ok=True)
+    seen = []
+    client = TestClient(Starlette(routes=[Mount("/replay", app=replay_app(_flow(seen), input="solar sails"))]))
+    page = client.get("/replay/").text
+    assert "PraisonAI Replay: Article" in page and '"runner"' in page
+    run_id = client.post("/replay/api/replay/run", json={"input": "comets"}).json()["run_id"]
+    for _ in range(200):
+        data = client.get(f"/replay/api/replay/runs/{run_id}/events?wait=1").json()
+        if data["run"] and data["events"][-1]["event_type"] == "session_end":
+            break
+        threading.Event().wait(0.02)  # time.sleep may be stubbed by the suite
+    assert seen == ["comets"]
+    assert data["run"]["status"] == "completed"
+    assert [e.get("node_id") for e in data["events"] if e["event_type"] == "step_end"] == ["step:research", "step:write"]
+
+
+def test_yaml_target_has_no_input_box():
+    _aiui_embed()
+    from starlette.testclient import TestClient
+
+    from praisonai.replay import replay_app
+
+    page = TestClient(replay_app("agents.yaml")).get("/").text
+    assert '"ask":false' in page and "agents.yaml" in page
@@ -52,3 +52,36 @@ def test_record_live_agent_then_export_html(traces_dir, tmp_path):
    spans = json.loads(otlp.output)["resourceSpans"][0]["scopeSpans"][0]["spans"]
    print("OTLP spans:", [s["name"] for s in spans])
    assert any(s["name"].startswith("invoke_agent") for s in spans)
+
+
+def test_run_button_runs_a_real_team_and_draws_it(traces_dir):
+    """replay_app(team): one Run click runs both agents, recorded live; a second Run runs them again."""
+    mod = pytest.importorskip("praisonaiui.features.replay")
+    if not hasattr(mod, "create_replay_app"):
+        pytest.skip("installed aiui predates create_replay_app")
+    import threading
+
+    from starlette.testclient import TestClient
+
+    from praisonaiagents import Agent, AgentTeam, Task
+    from praisonai.replay import replay_app
+
+    researcher = Agent(name="Researcher", instructions="Give 2 short facts.", llm="gpt-4o-mini")
+    writer = Agent(name="Writer", instructions="Write one sentence from the facts.", llm="gpt-4o-mini")
+    team = AgentTeam(agents=[researcher, writer], tasks=[
+        Task(name="research", description="2 facts about solar sails", agent=researcher, expected_output="2 facts"),
+        Task(name="write", description="One sentence using the facts", agent=writer, expected_output="a sentence"),
+    ])
+    traces_dir.mkdir(parents=True, exist_ok=True)
+    client = TestClient(replay_app(team))
+    for attempt in (1, 2):
+        run_id = client.post("/api/replay/run", json={"input": ""}).json()["run_id"]
+        for _ in range(600):
+            data = client.get(f"/api/replay/runs/{run_id}/events?wait=1").json()
+            if data["run"] and data["events"] and data["events"][-1]["event_type"] == "session_end":
+                break
+            threading.Event().wait(0.1)  # time.sleep may be stubbed by the suite
+        types = [e["event_type"] for e in data["events"]]
+        print(f"\nRun {attempt} ({run_id}):", types)
+        assert types.count("llm_response") >= 2, "both agents called the LLM"
+        assert [e["node_id"] for e in data["events"] if e["event_type"] == "task_end"] == ["task:research", "task:write"]