feat(replay): embeddable Replay app, Run box, Play, and a map that follows the cursor

- create_replay_app(run=None): the Replay view as a small ASGI app (no dashboard)
  to mount in any FastAPI/Starlette app; /?run=<id> pins one run (iframe-friendly).
  With run(input, run_id) it shows a Run box: POST /api/replay/run (JSON only,
  one run at a time) starts a run and the page follows it live.
- render(el, {base, run, runner, approve}) so the view works under a mount prefix.
- Map nodes show their state at the cursor: not reached / running / done / failed /
  skipped, and "×N" counts real runs instead of events (a step that ran once
  no longer reads "×2"). Scrubbing or Play rewinds the map with the cursor.
- ▶ Play steps through a run on its own.
- events?wait=1 answers a not-yet-started run without a 404.
- Node names stay whole (squeezed or clipped) so accessible names contain the
  visible text (WCAG 2.5.3; axe flagged truncated tool names).

Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
This commit is contained in:
MervinPraisonandClaude Opus 5.5 committed 2026-10-26 15:46:54 +00:00
1 parent bf98d29a38
commit 7037cf7bd3
7 files changed
+776 -50

No files matched your search

+78 -5
View File
@@ -32,13 +32,83 @@ aiui run "$(python -c 'import praisonaiui.replay_app as m; print(m.__file__)')"
`/#/replay?run=<id>` works too. In a full dashboard, Replay is in the sidebar under
**Control**.
## Watch a run live: the Run button
For learning, demos and debugging, open a page with a **Run** button. Each click records
a new run, and the page draws it as it happens:
```python
from praisonai.replay import watch
watch(flow, input="solar sails") # an Agent, AgentTeam, AgentFlow or agents.yaml
```
```bash
praisonai replay watch agents.yaml
```
On the **Map**, each step and agent shows where the run is: dashed means not reached,
blue (pulsing) means running now, green means done and red means failed. After a run,
**▶ Play** moves through it again one step at a time, and the map rewinds with it. One
run goes at a time. Each run of a team starts its tasks afresh; an Agent keeps its chat
history between runs.
## Show progress inside your own app
`create_replay_app()` is the same view as a small ASGI app, without the dashboard. Mount
it in any FastAPI or Starlette app:
```python
from praisonai.replay import record, replay_app
app.mount("/replay", replay_app()) # read-only: shows recorded runs
# wherever your app starts work:
with record(run_id, live=True):
team.start()
```
Then show one run, live, in your page, for example in an `<iframe>`:
```html
<iframe src="/replay/?run=RUN_ID" style="width:100%;height:640px"></iframe>
```
| URL | Shows |
|---|---|
| `/replay/` | The run list and the selected run |
| `/replay/?run=<id>` | Only that run, no list. A run that has not started yet is waited for. |
`replay_app(team)` adds the Run box to the mounted page. Without the PraisonAI wrapper,
pass any `run(input, run_id)` callable to `praisonaiui.features.replay.create_replay_app`;
it must record the run under `run_id`. For a page you build yourself, the view is one
module:
```html
<div id="progress"></div>
<script type="module">
import { render } from "/replay/replay.js";
render(document.getElementById("progress"), { base: "/replay", run: "RUN_ID", approve: false });
</script>
```
`render()` options: `base` (where the API is mounted), `run` (show only this run),
`runner` (`{input, placeholder, ask}` to show the Run box) and `approve` (`false` hides
the approve and deny buttons; the embedded app has no approvals API, so answer approvals
where your app asks for them).
Anyone who can reach the mounted app can read the runs, and with a runner can start one.
Keep it on localhost or behind your app's login. `POST /api/replay/run` only accepts
`Content-Type: application/json`, so another website cannot start a run from a visitor's
browser.
## What you see
| Part | What it shows |
|---|---|
| **Runs** | Recorded runs, newest first. A run that is still being written has a **Live** badge. |
| **Timeline** (default) | One lane per agent, plus **Manager** and **You**. Time runs left to right. LLM calls and tool calls are blocks in their agent's lane. Arrows show hand-offs, delegation, routing and approval requests. Parallel branches get their own sub-lanes. A call or step repeated back to back is one block marked ×N. Tasks and steps run across the top. |
| **Map** | The team or flow as a graph, drawn from the graph recorded at the start of the run and fitted to the page (−, Fit and + zoom; drag to pan). Agents that took part but are not in the graph, for example after a hand-off to an agent outside the team, appear in an **Outside the team** group with their hand-offs. Older recordings without a graph get a "who talked to whom" map built from the events. Nodes you have not reached yet have a dashed outline; the node the cursor is on has a bold one. |
| **Map** | The team or flow as a graph, drawn from the graph recorded at the start of the run and fitted to the page (−, Fit and + zoom; drag to pan). Agents that took part but are not in the graph, for example after a hand-off to an agent outside the team, appear in an **Outside the team** group with their hand-offs. Older recordings without a graph get a "who talked to whom" map built from the events. Each node shows its state at the cursor: dashed means not reached, blue means running, green means done (×N when it ran more than once) and red means failed; the node the cursor is on has a bold outline. |
| **What happened** | One plain sentence per step, for example *"Researcher asked web_search for 'solar sail facts' — 1.2 s, $0.002."* Hovering or focusing a sentence highlights its block, and the other way round. |
| **Details** | Three tabs for the selected step. **What happened**: when, who, how long, cost, tokens, output. **Why**: only what the recording holds, such as the input given to a tool, the context passed in a hand-off and its reason, the manager's task snapshot, the route a step chose, and an approval's decision and where an automatic approval came from. **Raw JSON**: the events themselves. |
| **Cost & latency** | Adds a bar under each block (taller and darker means more cost, or more time) and per-agent totals in the lane labels. |
@@ -65,7 +135,8 @@ anything after the cursor is drawn as an outline.
| Enter or Space | Open the focused block or node |
| ← → in the tabs | Switch between What happened, Why and Raw JSON |
The scrubber under the timeline does the same with a mouse or touch. On a live run the
The scrubber under the timeline does the same with a mouse or touch, and **▶ Play** steps
through on its own. On a live run the
cursor follows new events; if you step back, **Jump to live** takes you to the latest.
## Approving from Replay
@@ -92,11 +163,13 @@ an exported file.
| Endpoint | Returns |
|---|---|
| `GET /api/replay/runs` | `[{run_id, session_id, path, started_at, ended_at?, status, kind, name, agents, events, cost_usd, live}]`, newest first |
| `GET /api/replay/runs/{id}/events?after_seq=N` | `{events, graph, run}`; `after_seq` keeps only events with a larger `sequence_num` |
| `GET /api/replay/runs/{id}/events?after_seq=N` | `{events, graph, run}`; `after_seq` keeps only events with a larger `sequence_num`; `wait=1` answers `{events: [], graph: null, run: null}` instead of 404 for a run that has not started yet |
| `GET /api/replay/runs/{id}/stream` | Server-sent events: `event: ctx` per trace event (with `id:` = `sequence_num`), `ping` every 15 s, `end` once the run has finished and gone quiet for 2 s. Honours `after_seq` and `Last-Event-ID`. |
Run ids must match `^[A-Za-z0-9._-]+$`. Replay is read-only: it never writes trace
files. Half-written last lines, unknown fields and unknown event types are skipped
With a runner (`create_replay_app(run=...)`), `POST /api/replay/run` with
`{"input": "..."}` starts a run and returns `{run_id}` (409 while one is going).
Run ids must match `^[A-Za-z0-9._-]+$`. Replay never writes trace files itself. Half-written last lines, unknown fields and unknown event types are skipped
safely, and schema 1.0 recordings still open.
The **Traces** page reads the same files: each run is a trace, and its spans come from
+145 -3
View File
@@ -3,7 +3,9 @@
Routes (contract v1, section 5):
* ``GET /api/replay/runs`` run picker rows, newest first
* ``GET /api/replay/runs/{run_id}/events`` ``{events, graph, run}``; ``?after_seq=N``
* ``GET /api/replay/runs/{run_id}/events`` ``{events, graph, run}``; ``?after_seq=N``;
``?wait=1`` answers ``{events: [], graph: null, run: null}`` (not 404) for a run
that has not written its first event yet
* ``GET /api/replay/runs/{run_id}/stream`` SSE tail of the JSONL file
The trace directory is ``$PRAISONAI_TRACES_DIR``, else ``$PRAISON_HOME/traces``,
@@ -12,20 +14,30 @@ else ``~/.praison/traces``. Run ids must match ``^[A-Za-z0-9._-]+$``.
The browser side lives in ``templates/frontend/plugins/views/replay-core.js``
(pure functions) and ``replay.js`` (DOM/SVG). :func:`standalone_html` inlines
both into one self-contained page for ``praisonai replay export --format html``.
:func:`create_replay_app` is the same view as a small ASGI app (no dashboard) to
mount inside your own app, optionally with a Run box that starts runs::
app.mount("/replay", create_replay_app(run=my_runner))
"""
from __future__ import annotations
import asyncio
import html
import inspect
import json
import logging
import threading
import time
import uuid
from pathlib import Path
from typing import Any, Dict, List, Optional
from typing import Any, Callable, Dict, List, Optional
from urllib.parse import quote
from starlette.concurrency import run_in_threadpool
from starlette.requests import Request
from starlette.responses import JSONResponse, StreamingResponse
from starlette.responses import FileResponse, HTMLResponse, JSONResponse, StreamingResponse
from starlette.routing import Route
from . import _trace_jsonl as tj
@@ -233,6 +245,8 @@ class ReplayFeature(BaseFeatureProtocol):
async def _events(self, request: Request) -> JSONResponse:
path, err = _resolve(request)
if err is not None:
if err.status_code == 404 and request.query_params.get("wait") == "1":
return JSONResponse({"events": [], "graph": None, "run": None})
return err
after_seq, err = _parse_after_seq(request)
if err is not None:
@@ -297,3 +311,131 @@ class ReplayFeature(BaseFeatureProtocol):
media_type="text/event-stream",
headers={"Cache-Control": "no-cache", "Connection": "keep-alive", "X-Accel-Buffering": "no"},
)
# ── embeddable app ─────────────────────────────────────────────────────
#: Longest input the Run box accepts (characters).
MAX_RUN_INPUT = 20_000
def _embed_page(base: str, title: str, cfg: Dict[str, Any]) -> str:
"""The page at the app's root: the view, mounted with *cfg*, and nothing else."""
return (
"<!doctype html>\n<html lang=\"en\">\n<head>\n<meta charset=\"utf-8\">\n"
"<meta name=\"viewport\" content=\"width=device-width, initial-scale=1\">\n"
f"<title>{html.escape(title)}</title>\n"
f"<link rel=\"icon\" href=\"data:image/svg+xml,{quote(REPLAY_MARK_SVG.replace('currentColor', '#1A56DB'))}\">\n"
"<style>body{margin:0;font-family:system-ui,-apple-system,'Segoe UI',sans-serif}"
"#replay-root{padding:12px;box-sizing:border-box;max-width:100%}</style>\n"
"</head>\n<body>\n<main id=\"replay-root\"></main>\n"
"<script type=\"module\">\n"
f"import {{ render }} from {_json_for_script(base + '/replay.js')};\n"
f"render(document.getElementById('replay-root'), {_json_for_script(cfg)});\n"
"</script>\n</body>\n</html>\n"
)
def create_replay_app(
run: Optional[Callable[[str, str], Any]] = None,
*,
title: str = "PraisonAI Replay",
input: Optional[str] = "",
placeholder: str = "",
):
"""PraisonAI Replay as a small ASGI app to mount in your own app.
Serves the view at ``/`` (no dashboard around it), its two scripts, and the
``/api/replay/...`` routes, all relative to where it is mounted::
from fastapi import FastAPI
from praisonaiui.features.replay import create_replay_app
app = FastAPI()
app.mount("/replay", create_replay_app())
# <iframe src="/replay/?run=RUN_ID"> shows one run live; /replay/ lists runs
Args:
run: Optional ``run(input, run_id)``. When given, the page shows an
input box and a **Run** button; each click calls it in a worker
thread with a new ``run_id``, and the page follows that run live.
It must record the run under ``run_id`` (``praisonai.replay.app()``
does this for an agent, team or flow). One run at a time.
title: Page title.
input: Text the input box starts with; ``None`` for no input box
(just a Run button, e.g. for a YAML file that has its own topic).
placeholder: Placeholder text for the input box.
Traces are read from the same folder as the rest of Replay
(``$PRAISONAI_TRACES_DIR`` > ``$PRAISON_HOME/traces`` > ``~/.praison/traces``).
Anyone who can reach the app can read the runs (and, with ``run``, start
them): keep it on localhost or behind your app's auth.
"""
from starlette.applications import Starlette
feature = ReplayFeature()
busy = threading.Lock()
async def page(request: Request) -> HTMLResponse:
base = str(request.scope.get("root_path", "")).rstrip("/")
cfg: Dict[str, Any] = {"base": base, "approve": False}
wanted = request.query_params.get("run")
if wanted and tj.is_valid_run_id(wanted):
cfg["run"] = wanted
if run is not None:
cfg["runner"] = {"input": input or "", "placeholder": placeholder, "ask": input is not None}
return HTMLResponse(_embed_page(base, title, cfg), headers={"Cache-Control": "no-store"})
def script(path: Path):
async def serve(request: Request) -> FileResponse:
return FileResponse(path, media_type="text/javascript", headers={"Cache-Control": "no-cache"})
return serve
def _call(text: str, run_id: str) -> None:
try:
result = run(text, run_id) # type: ignore[misc]
if inspect.isawaitable(result):
asyncio.run(result)
except Exception:
logger.exception("replay run %s failed", run_id)
finally:
busy.release()
async def start(request: Request) -> JSONResponse:
# JSON only: a cross-site form post cannot set this content type without
# a CORS preflight, which this app never answers.
if not request.headers.get("content-type", "").lower().startswith("application/json"):
return JSONResponse({"error": "send JSON: {\"input\": \"...\"}"}, status_code=415)
try:
body = await request.json()
except ValueError:
return _bad_request("body must be JSON")
text = body.get("input", "") if isinstance(body, dict) else None
if not isinstance(text, str) or len(text) > MAX_RUN_INPUT:
return _bad_request(f"input must be a string of at most {MAX_RUN_INPUT} characters")
if not busy.acquire(blocking=False):
return JSONResponse({"error": "a run is already going; wait for it to finish"}, status_code=409)
run_id = f"run-{uuid.uuid4().hex[:12]}"
try:
threading.Thread(target=_call, args=(text, run_id), name=f"replay-{run_id}", daemon=True).start()
except Exception:
busy.release()
raise
# Answer once the trace file exists (usually milliseconds), so the page
# can open the run straight away; it waits on its own if this times out.
path = tj.run_path(run_id)
for _ in range(40):
if path is not None and path.is_file():
break
await asyncio.sleep(0.05)
return JSONResponse({"run_id": run_id})
routes = [
Route("/", page, methods=["GET"]),
Route("/replay.js", script(REPLAY_VIEW_JS), methods=["GET"]),
Route("/replay-core.js", script(REPLAY_CORE_JS), methods=["GET"]),
*feature.routes(),
]
if run is not None:
routes.append(Route("/api/replay/run", start, methods=["POST"]))
return Starlette(routes=routes)
@@ -1024,6 +1024,56 @@ export function layoutRunMap(graph, events, opts = {}) {
};
}
/** End events that mean "this node finished one run", and which node that is. */
const RUN_ENDS = {
step_end: (nid, ev) => nid === ev.node_id,
task_end: (nid, ev) => nid === ev.node_id,
agent_end: (nid) => nid.startsWith('agent:') || nid === MANAGER_LANE,
tool_call_end: (nid) => nid.startsWith('tool:'),
};
/**
* Each map node's state after the first ``upTo + 1`` events (sorted, as in
* buildModel().events): ``{events, runs, state}`` where state is "running"
* (started, not ended yet), "done", "failed" or "skipped" (from the last end's
* ``data.status``). Nodes without an entry were not reached by then. The flow's
* start/end nodes count as reached once the run started/ended.
*/
export function nodeStates(events, upTo = Infinity, nodes = []) {
const out = new Map();
const last = Math.min(events.length - 1, upTo);
let ended = false;
for (let i = 0; i <= last; i++) {
const ev = events[i];
const type = ev && ev.event_type;
if (type === 'workflow_end' || type === 'session_end') ended = true;
for (const nid of nodeIdsForEvent(ev)) {
const st = out.get(nid) || { events: 0, runs: 0, open: 0, last: null };
st.events += 1;
if (PAIRS[type]) st.open += 1;
else if (END_OF[type]) {
st.open = Math.max(0, st.open - 1);
if (RUN_ENDS[type] && RUN_ENDS[type](nid, ev)) {
st.runs += 1;
const status = str(dataOf(ev).status);
st.last = status === 'failed' || dataOf(ev).error ? 'failed' : (status === 'skipped' ? 'skipped' : 'done');
}
}
out.set(nid, st);
}
}
for (const n of nodes || []) {
if (out.has(n.id)) continue;
if ((n.kind === 'start' && last >= 0) || (n.kind === 'end' && ended)) out.set(n.id, { events: 1, runs: 1, open: 0, last: 'done' });
}
for (const st of out.values()) {
st.state = st.open > 0 ? 'running' : (st.last || 'done');
delete st.open;
delete st.last;
}
return out;
}
/** Words for edge kinds on the map (labels never rely on colour). */
export const EDGE_WORDS = Object.freeze({
handoff: 'hands off',
@@ -1045,7 +1095,7 @@ export function edgeText(edge) {
const api = {
PALETTE, SHAPES, HUMAN_LANE, MANAGER_LANE, EDGE_WORDS,
buildModel, narrate, layoutMap, layoutRunMap, observedGraph, deriveGraph, nodeIdsForEvent, pairEvents, sortEvents,
buildModel, narrate, layoutMap, layoutRunMap, observedGraph, deriveGraph, nodeIdsForEvent, nodeStates, pairEvents, sortEvents,
eventId, formatDuration, formatCost, argPreview, truncate, stripPrefix, edgeText,
};
if (typeof globalThis !== 'undefined') globalThis.PraisonReplayCore = api;
@@ -15,6 +15,9 @@
*
* Standalone: with window.__REPLAY_DATA__ = {events, graph} set and no aiui
* dashboard present, the module renders itself into #replay-root (or <body>).
*
* Embedded (any page): render(el, {base, run, runner}) — see render() below and
* praisonaiui.features.replay.create_replay_app() for the server side.
* Pure logic lives in replay-core.js.
*/
@@ -131,6 +134,12 @@ const STYLE = `
.rp-btn[disabled]{opacity:.55;cursor:not-allowed}
.rp-btn-primary{background:var(--rp-human);border-color:var(--rp-human);color:var(--rp-on-fill)}
.rp-seg{display:inline-flex;gap:0}.rp-seg .rp-btn{border-radius:0}.rp-seg .rp-btn:first-child{border-radius:8px 0 0 8px}.rp-seg .rp-btn:last-child{border-radius:0 8px 8px 0;margin-left:-1px}
.rp-runbox{display:flex;gap:8px;align-items:center;flex-wrap:wrap;padding:10px 14px;border-bottom:1px solid var(--rp-border)}
.rp-runbox[hidden]{display:none}
.rp-input{flex:1;min-width:180px;font:inherit;font-size:14px;color:var(--rp-text);background:var(--rp-bg);border:1px solid var(--rp-control);border-radius:8px;padding:6px 10px;min-height:34px}
.rp-primary{background:var(--rp-focus);border-color:var(--rp-focus);color:var(--rp-bg);font-weight:600;min-height:34px;padding:4px 16px}
.rp-btn:disabled{opacity:.6;cursor:default}
.rp-primary:disabled{opacity:1;background:var(--rp-lane);border-color:var(--rp-control);color:var(--rp-text)}
.rp-select{font:inherit;font-size:13px;color:var(--rp-text);background:var(--rp-bg);border:1px solid var(--rp-control);border-radius:8px;padding:3px 6px;min-height:30px}
.rp-pending{display:flex;flex-direction:column;gap:8px;padding:10px 14px;border-bottom:1px solid var(--rp-border);background:var(--rp-surface)}
.rp-pending[hidden]{display:none}
@@ -224,6 +233,10 @@ const STYLE = `
.rp-svg .rp-node .rp-card{fill:var(--rp-bg);stroke:var(--rp-control);stroke-width:1.5}
.rp-svg .rp-node.rp-unvisited .rp-card{stroke-dasharray:4 3}
.rp-svg .rp-node.rp-now .rp-card{stroke:var(--rp-text);stroke-width:3}
.rp-svg .rp-node.rp-st-running .rp-card{stroke:var(--rp-focus);stroke-width:3}
.rp-svg .rp-node.rp-st-done .rp-card{stroke:var(--rp-ok)}
.rp-svg .rp-node.rp-st-failed .rp-card{stroke:var(--rp-err);stroke-width:2.5}
.rp-svg .rp-node.rp-st-skipped .rp-card{opacity:.6}
.rp-svg .rp-node{cursor:pointer}
.rp-svg .rp-node .rp-kindword{fill:var(--rp-dim);font-size:11px}
.rp-svg .rp-edge path{fill:none;stroke:var(--rp-control);stroke-width:1.6}
@@ -237,7 +250,8 @@ const STYLE = `
.rp-svg .rp-group rect{fill:none;stroke:var(--rp-control);stroke-width:1.5;stroke-dasharray:6 4}
.rp-map{cursor:grab}.rp-map.rp-panning{cursor:grabbing}
@media (prefers-reduced-motion:no-preference){
.rp-b-live .rp-dot{animation:rp-pulse 1.6s ease-in-out infinite}
.rp-b-live .rp-dot,.rp-svg .rp-node.rp-st-running .rp-ring{animation:rp-pulse 1.6s ease-in-out infinite}
.rp-svg .rp-node.rp-st-running .rp-ring{stroke:var(--rp-focus);stroke-width:2}
.rp-step,.rp-svg .rp-block,.rp-svg .rp-arrow{transition:opacity .15s ease}
}
@keyframes rp-pulse{0%,100%{opacity:1}50%{opacity:.35}}
@@ -288,7 +302,7 @@ function runFromUrl() {
}
function writeRunToUrl(id) {
if (S.solo) return;
if (S.solo || S.pinned) return;
try {
const url = new URL(window.location.href);
url.searchParams.set('run', id);
@@ -326,6 +340,14 @@ function statusBadge(row) {
/**
* Mount the view. Dashboard: render(container). Standalone:
* render(container, {standalone: true, events, graph, title}).
*
* Embedded in another page (options are all optional):
* base — URL prefix the replay API is mounted under, e.g. "/replay"
* run — show only this run id (no run list); waits for it to start
* runner — {input, placeholder, ask}: show a Run box that POSTs {input} to
* <base>/api/replay/run and follows the run it returns
* (ask: false shows just the Run button)
* approve — false to hide the approve/deny buttons (no approvals API here)
*/
export async function render(container, opts = {}) {
cleanup();
@@ -335,6 +357,11 @@ export async function render(container, opts = {}) {
S = {
core,
solo,
base: String(opts.base || '').replace(/\/+$/, ''),
pinned: opts.run ? String(opts.run) : null,
runner: opts.runner && typeof opts.runner === 'object' ? { input: '', ...opts.runner } : null,
canApprove: !solo && opts.approve !== false,
play: null,
container,
abort: new AbortController(),
timers: [],
@@ -369,6 +396,8 @@ export async function render(container, opts = {}) {
setEvents(opts.events || [], opts.graph || null);
S.cursor = S.model.narration.length - 1;
renderAll();
} else if (S.pinned) {
await loadRun(S.pinned);
} else {
await refreshRuns();
const wanted = runFromUrl();
@@ -400,6 +429,7 @@ export function cleanup() {
try { S.abort.abort(); } catch { /* ignore */ }
if (S.es) { try { S.es.close(); } catch { /* ignore */ } }
for (const t of S.timers) clearInterval(t);
if (S.play) clearInterval(S.play);
if (S.ro) { try { S.ro.disconnect(); } catch { /* ignore */ } }
if (S.unlisten) { try { S.unlisten(); } catch { /* ignore */ } }
S = null;
@@ -409,14 +439,20 @@ export function cleanup() {
async function fetchJson(url, init = {}) {
const res = await fetch(url, { signal: S.abort.signal, ...init });
if (!res.ok) throw new Error(`HTTP ${res.status}`);
if (!res.ok) {
let msg = `HTTP ${res.status}`;
try { const body = await res.json(); if (body && body.error) msg = body.error; } catch { /* not JSON */ }
const err = new Error(msg);
err.status = res.status;
throw err;
}
return res.json();
}
async function refreshRuns(autoload = false) {
if (!S || S.solo) return;
if (!S || S.solo || S.pinned) return;
try {
const rows = await fetchJson('/api/replay/runs');
const rows = await fetchJson(`${S.base}/api/replay/runs`);
S.runs = Array.isArray(rows) ? rows : [];
} catch (e) {
if (e.name === 'AbortError') return;
@@ -482,15 +518,35 @@ async function loadRun(id) {
renderRunList();
let data;
try {
data = await fetchJson(`/api/replay/runs/${encodeURIComponent(id)}/events`);
const waiting = S.pinned === id || S.awaiting === id;
data = await fetchJson(`${S.base}/api/replay/runs/${encodeURIComponent(id)}/events${waiting ? '?wait=1' : ''}`);
if (waiting && !data.run) {
const err = new Error('run not found');
err.status = 404;
throw err;
}
} catch (e) {
if (e.name === 'AbortError') return;
if (e.status === 404 && (S.pinned === id || S.awaiting === id) && (S.waitTries = (S.waitTries || 0) + 1) <= 120) {
// the run was asked for but its first event is not written yet (wait up to 60 s)
S.run = { run_id: id, name: 'Starting…', status: 'running', live: true };
setEvents([], null);
renderAll();
setTimeout(() => { if (S && S.runId === id) loadRun(id); }, 500);
return;
}
S.waitTries = 0;
S.run = { run_id: id, name: id, status: 'incomplete', live: false };
setEvents([], null);
renderAll(`Could not load run ${id} (${e.message}).`);
return;
}
if (!S || S.runId !== id) return;
S.waitTries = 0;
if (S.awaiting === id) {
S.awaiting = null;
refreshRuns(); // replace the "Starting…" row with the real one
}
S.run = data.run || S.runs.find((r) => r.run_id === id) || { run_id: id, name: id };
S.follow = true;
setEvents(data.events || [], data.graph || null);
@@ -508,7 +564,7 @@ function maxSeq() {
function startStream(id) {
if (typeof EventSource === 'undefined') return;
const after = maxSeq();
const url = `/api/replay/runs/${encodeURIComponent(id)}/stream${after != null ? `?after_seq=${after}` : ''}`;
const url = `${S.base}/api/replay/runs/${encodeURIComponent(id)}/stream${after != null ? `?after_seq=${after}` : ''}`;
const es = new EventSource(url);
S.es = es;
es.addEventListener('ctx', (e) => {
@@ -533,8 +589,8 @@ function buildShell() {
clear(c);
const els = {};
els.root = h('section', { class: 'rp-root', 'data-theme': S.theme, 'aria-label': 'PraisonAI Replay' });
els.shell = h('div', { class: `rp-shell${S.solo ? ' rp-solo' : ''}` });
if (!S.solo) {
els.shell = h('div', { class: `rp-shell${S.solo || S.pinned ? ' rp-solo' : ''}` });
if (!S.solo && !S.pinned) {
els.runs = h('nav', { class: 'rp-runs', 'aria-label': 'Recorded runs' });
els.shell.appendChild(els.runs);
}
@@ -549,7 +605,8 @@ function buildShell() {
els.scrub = h('div', { class: 'rp-scrub' });
els.drawer = h('div', { class: 'rp-drawer', role: 'region', 'aria-label': 'Details' });
els.live = h('div', { class: 'rp-sr', 'aria-live': 'polite', role: 'status' });
els.main.append(els.bar, els.pending, els.stage, els.legend, els.scrub, els.drawer);
els.runbox = h('form', { class: 'rp-runbox', hidden: true, onsubmit: (e) => { e.preventDefault(); startRun(); } });
els.main.append(els.runbox, els.bar, els.pending, els.stage, els.legend, els.scrub, els.drawer);
els.shell.appendChild(els.main);
els.root.append(els.shell, els.live);
els.root.addEventListener('keydown', onKeyDown);
@@ -600,6 +657,7 @@ function renderAll(errorText) {
S.els.root.setAttribute('data-theme', S.theme);
renderRunList();
renderBar();
renderRunBox();
if (!S.solo && !S.runs.length && !S.runId) {
renderEmpty();
return;
@@ -619,6 +677,12 @@ function renderEmpty() {
for (const k of ['pending', 'narr', 'scrub', 'drawer', 'legend']) clear(S.els[k]);
S.els.pending.hidden = true;
clear(S.els.canvas);
if (S.runner) {
S.els.canvas.appendChild(h('div', { class: 'rp-empty' },
h('h2', { text: 'Nothing has run yet' }),
h('p', { text: 'Type an input above and press Run. Each step lights up on the map as it happens.' })));
return;
}
S.els.canvas.appendChild(h('div', { class: 'rp-empty' },
h('h2', { text: 'No recorded runs yet' }),
h('p', { text: 'Replay shows who did what, when, why and at what cost, once a run has been recorded. Record one and it appears here (live runs update as they go):' }),
@@ -629,6 +693,56 @@ function renderEmpty() {
));
}
// ── run box (embedded with a runner) ──────────────────────────────────
/** Input + Run button; built once so typing survives live re-renders. */
function renderRunBox() {
const box = S.els.runbox;
if (!S.runner || !box) return;
box.hidden = false;
if (!box.firstChild) {
append(box, [
S.runner.ask === false ? null : h('label', { for: 'rp-run-input', class: 'rp-sr', text: 'Input for the run' }),
h('input', { id: 'rp-run-input', class: 'rp-input', type: S.runner.ask === false ? 'hidden' : 'text', value: S.runner.input || '', placeholder: S.runner.placeholder || 'What should the agents work on?', 'data-key': 'run-input' }),
h('button', { type: 'submit', class: 'rp-btn rp-primary', 'data-key': 'run' }),
h('span', { class: 'rp-note rp-runbox-msg', role: 'status' })]);
}
const busy = Boolean(S.awaiting || S.starting || (S.run && S.run.live));
const btn = box.querySelector('button');
btn.disabled = busy;
btn.textContent = busy ? 'Running…' : '▶ Run';
}
async function startRun() {
if (!S || !S.runner || S.starting) return;
const box = S.els.runbox;
const msg = box.querySelector('.rp-runbox-msg');
msg.textContent = '';
S.starting = true;
renderRunBox();
let res;
try {
res = await fetchJson(`${S.base}/api/replay/run`, {
method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ input: box.querySelector('input').value }),
});
} catch (e) {
if (e.name === 'AbortError' || !S) return;
S.starting = false;
msg.textContent = `Could not start the run: ${e.message}`;
renderRunBox();
return;
}
if (!S) return;
S.starting = false;
S.awaiting = res.run_id;
S.waitTries = 0;
stopPlay();
if (S.pinned) S.pinned = res.run_id;
else S.runs = [{ run_id: res.run_id, name: 'Starting…', status: 'running', live: true, events: 0, agents: [] }, ...S.runs.filter((r) => r.run_id !== res.run_id)];
announce('Run started.');
await loadRun(res.run_id);
}
// ── run picker ────────────────────────────────────────────────────────
function renderRunList() {
@@ -1224,19 +1338,30 @@ function mapGraph() {
return mapLayout().graph;
}
function nodeVisits() {
const counts = new Map();
/** Event index the cursor is on (-1 before the first step). */
function cursorEventIndex() {
const m = S.model;
for (const ev of m.events) {
for (const nid of C().nodeIdsForEvent(ev)) counts.set(nid, (counts.get(nid) || 0) + 1);
}
// Brief/Delivered are reached when the run starts/ends, though no event names them.
const ended = m.events.some((e) => e.event_type === 'workflow_end' || e.event_type === 'session_end');
for (const n of mapGraph().nodes || []) {
if (n.kind === 'start' && m.events.length && !counts.has(n.id)) counts.set(n.id, 1);
if (n.kind === 'end' && ended && !counts.has(n.id)) counts.set(n.id, 1);
}
return { counts };
const item = m.narration[S.cursor];
if (!item) return -1;
const i = m.eventIndex.get(item.id);
return typeof i === 'number' ? i : m.events.length - 1;
}
/** Each node's state at the cursor (see replay-core nodeStates). */
function nodeStatesAtCursor() {
return C().nodeStates(S.model.events, cursorEventIndex(), mapGraph().nodes || []);
}
const STATE_WORD = { running: 'running now', done: 'done', failed: 'failed', skipped: 'skipped' };
/** What a node did up to the cursor: "done ×2" on the card (its shape shows the kind), full words for screen readers. */
function nodeStateText(n, st) {
const word = KIND_WORD[n.kind] || n.kind;
if (!st) return { line: word, label: `${n.label} ${word}, not reached yet` };
const times = st.runs > 1 ? ` ×${st.runs}` : '';
const state = STATE_WORD[st.state] || st.state;
// the accessible name starts with the visible text (WCAG 2.5.3 label in name)
return { line: `${state}${times}`, label: `${n.label} ${state}${times} (${word}${st.runs > 1 ? `, ran ${st.runs} times` : ''})` };
}
function shapePath(shape, x, y, w, h) {
@@ -1263,6 +1388,26 @@ function nodeGlyph(node, gx, gy) {
const MAP_MIN_SCALE = 0.55; // below this, labels stop being readable: scroll instead
/**
* A node's name, whole (so its accessible name contains the visible text,
* WCAG 2.5.3): squeezed a little when slightly too long, clipped at the card's
* edge when much longer (the <title> and the drawer show it in full).
*/
function nodeLabel(n, i, svg) {
const room = n.w - 52;
const need = String(n.label).length * 7.2;
const attrs = { x: n.x + 44, y: n.y + n.h / 2 - 2, 'font-weight': '600', 'aria-hidden': 'true', text: n.label };
if (need > room && need <= room * 1.3) {
attrs.textLength = room;
attrs.lengthAdjust = 'spacingAndGlyphs';
} else if (need > room) {
const id = `rp-clip-${i}`;
svg.querySelector('defs').appendChild(s('clipPath', { id }, s('rect', { x: n.x + 40, y: n.y, width: room + 4, height: n.h })));
attrs['clip-path'] = `url(#${id})`;
}
return s('text', attrs);
}
function renderMap() {
const canvas = S.els.canvas;
const prev = { left: canvas.scrollLeft, top: canvas.scrollTop };
@@ -1274,7 +1419,6 @@ function renderMap() {
canvas.appendChild(h('div', { class: 'rp-empty' }, h('h2', { text: 'Nothing to map yet' }), h('p', { text: 'The map appears once agents start talking.' })));
return;
}
const visits = nodeVisits();
const W = lay.width + 20;
const H = lay.height + 20;
const cw = Math.max(200, (canvas.clientWidth || 900) - 8);
@@ -1318,14 +1462,9 @@ function renderMap() {
const groups = { main: s('g', { role: 'group', 'aria-label': 'Nodes' }), outside: null };
if ((lay.groups || []).length) groups.outside = s('g', { role: 'group', 'aria-label': lay.groups[0].label });
ordered.forEach((n, i) => {
const count = visits.counts.get(n.id) || 0;
const word = KIND_WORD[n.kind] || n.kind;
const kindLine = count ? `${word} · ×${count}` : word;
const name = `${n.label} ${kindLine}${count ? `, ${count} events here` : ', not reached yet'}`;
const g = s('g', {
class: `rp-node${count ? '' : ' rp-unvisited'}${S.nodeSel === n.id ? ' rp-sel' : ''}`,
class: `rp-node${S.nodeSel === n.id ? ' rp-sel' : ''}`,
'data-node': n.id, 'data-key': `node:${n.id}`, role: 'button', tabindex: i === 0 ? '0' : '-1',
'aria-label': name,
onclick: () => selectNode(n.id),
onfocus: () => highlight(null),
});
@@ -1334,9 +1473,9 @@ function renderMap() {
g.appendChild(path ? s('path', { class: 'rp-card', d: path }) : s('rect', { class: 'rp-card', x: n.x, y: n.y, width: n.w, height: n.h, rx: n.kind === 'tool' ? 14 : (n.kind === 'end' ? 2 : 6) }));
if (n.kind === 'end') g.appendChild(s('rect', { class: 'rp-card', x: n.x + 3, y: n.y + 3, width: n.w - 6, height: n.h - 6, rx: 2, fill: 'none' }));
g.appendChild(nodeGlyph(n, n.x + 12, n.y + n.h / 2 - 11));
g.appendChild(s('text', { x: n.x + 44, y: n.y + n.h / 2 - 2, 'font-weight': '600', 'aria-hidden': 'true', text: fitLabel(n.label, n.w - 52, 7.2) }));
g.appendChild(s('text', { class: 'rp-kindword', x: n.x + 44, y: n.y + n.h / 2 + 13, 'aria-hidden': 'true', text: ` ${kindLine}` }));
g.appendChild(s('title', { text: name }));
g.appendChild(nodeLabel(n, i, svg));
g.appendChild(s('text', { class: 'rp-kindword', x: n.x + 44, y: n.y + n.h / 2 + 13, 'aria-hidden': 'true' }));
g.appendChild(s('title'));
(n.outside && groups.outside ? groups.outside : groups.main).appendChild(g);
});
svg.appendChild(groups.main);
@@ -1367,12 +1506,25 @@ function enablePan(svg, box) {
svg.addEventListener('pointercancel', end);
}
/** Node states follow the cursor: not reached (dashed), running, done, failed. */
function applyMapCursor() {
const item = S.model.narration[S.cursor];
const ev = item ? S.model.events[S.model.eventIndex.get(item.id)] : null;
const ev = S.model.events[cursorEventIndex()];
const now = new Set(ev ? C().nodeIdsForEvent(ev) : []);
const states = nodeStatesAtCursor();
const byId = new Map((mapLayout().nodes || []).map((n) => [n.id, n]));
for (const g of S.els.canvas.querySelectorAll('.rp-node')) {
g.classList.toggle('rp-now', now.has(g.getAttribute('data-node')));
const id = g.getAttribute('data-node');
const st = states.get(id);
const n = byId.get(id) || { id, label: id, kind: '' };
const { line, label } = nodeStateText(n, st);
g.classList.toggle('rp-now', now.has(id));
g.classList.toggle('rp-unvisited', !st);
for (const k of ['running', 'done', 'failed', 'skipped']) g.classList.toggle(`rp-st-${k}`, Boolean(st && st.state === k));
g.setAttribute('aria-label', label);
const kw = g.querySelector('.rp-kindword');
if (kw) kw.textContent = ` ${fitLabel(line, (n.w || 136) - 52, 6.2)}`;
const tip = g.querySelector('title');
if (tip) tip.textContent = label;
}
}
@@ -1403,7 +1555,7 @@ function renderLegend() {
lg.append(h('span', { text: 'solid arrow: hand-off · dashed: delegate · dotted: route' }), h('span', { text: 'outlined block: not reached yet' }));
if (S.overlay) lg.append(h('span', { text: `bar under a block: ${S.heat === 'cost' ? 'cost' : 'latency'} (taller and darker = more)` }));
} else {
lg.append(h('span', { text: 'dashed outline: not reached · bold outline: now' }));
lg.append(h('span', { text: 'dashed: not reached · blue: running · green: done · red: failed · bold: the step you are on' }));
}
}
@@ -1447,6 +1599,11 @@ function renderScrubber() {
const item = S.model.narration[S.cursor];
const existing = box.querySelector('input[type=range]');
const valueText = item ? `Step ${S.cursor + 1} of ${n}: ${item.text}` : 'No steps';
const playBtn = box.querySelector('[data-key="play"]');
if (playBtn) {
playBtn.textContent = S.play ? '❚❚ Pause' : '▶ Play';
playBtn.setAttribute('aria-pressed', S.play ? 'true' : 'false');
}
if (existing && Number(existing.max) === Math.max(0, n - 1)) {
existing.value = String(Math.max(0, S.cursor));
existing.setAttribute('aria-valuetext', valueText);
@@ -1456,6 +1613,7 @@ function renderScrubber() {
clear(box);
if (!n) return;
box.append(
h('button', { type: 'button', class: 'rp-btn', 'data-key': 'play', 'aria-pressed': S.play ? 'true' : 'false', title: 'Play the run step by step', onclick: togglePlay }, S.play ? '❚❚ Pause' : '▶ Play'),
h('button', { type: 'button', class: 'rp-btn', 'data-key': 'prev', onclick: () => setCursor(S.cursor - 1) }, h('span', { 'aria-hidden': 'true' }, '← '), 'Previous'),
h('label', { for: 'rp-scrub-range', class: 'rp-sr', text: 'Step' }),
h('input', {
@@ -1469,6 +1627,28 @@ function renderScrubber() {
);
}
/** Play: move the cursor one step every PLAY_MS, from the start if at the end. */
const PLAY_MS = 900;
function togglePlay() {
if (!S) return;
if (S.play) { stopPlay(); return; }
if (S.cursor >= S.model.narration.length - 1) setCursor(0);
S.play = setInterval(() => {
if (!S || !S.model) return;
if (S.cursor >= S.model.narration.length - 1) { stopPlay(); return; }
setCursor(S.cursor + 1);
}, PLAY_MS);
renderScrubber();
}
function stopPlay() {
if (!S || !S.play) return;
clearInterval(S.play);
S.play = null;
renderScrubber();
}
// ── drawer ────────────────────────────────────────────────────────────
const TABS = [['what', 'What happened'], ['why', 'Why'], ['raw', 'Raw JSON']];
@@ -1702,7 +1882,7 @@ function renderPending() {
const pending = S.model.pending;
// Rebuild only when something visible changed, so a live update never
// resets the caret of someone typing arguments or a reason.
const sig = `${S.solo}|${Boolean(S.run?.live)}|${pending.map((b) => {
const sig = `${S.canApprove}|${Boolean(S.run?.live)}|${pending.map((b) => {
const st = S.asks.get(b.approvalId) || {};
return `${b.approvalId}:${Boolean(st.editing)}:${st.sent || ''}:${Boolean(st.loaded)}`;
}).join(',')}`;
@@ -1724,8 +1904,8 @@ function renderPending() {
`${req ? req.agent_name || 'An agent' : 'An agent'} asks to ${d.kind === 'task_output_review' ? 'deliver the output of' : 'run'} ${b.label}`,
d.risk ? h('span', { class: 'rp-badge', text: `${d.risk} risk` }) : null));
card.appendChild(h('pre', { tabindex: '0', 'aria-label': 'Recorded arguments (redacted)', text: prettyJson(d.args || {}) }));
if (S.solo) {
card.appendChild(h('p', { class: 'rp-note', text: 'This export cannot answer. Open the run with praisonai replay view to approve or deny.' }));
if (!S.canApprove) {
card.appendChild(h('p', { class: 'rp-note', text: S.solo ? 'This export cannot answer. Open the run with praisonai replay view to approve or deny.' : 'Answer this request where your app asks for approval.' }));
box.appendChild(card);
continue;
}
@@ -1792,7 +1972,7 @@ async function startEdit(b, recorded) {
st.text = prettyJson(recorded.args || {});
st.loaded = false;
try {
const res = await fetch(`/api/approvals/${encodeURIComponent(b.approvalId)}`, { signal: S.abort.signal });
const res = await fetch(`${S.base}/api/approvals/${encodeURIComponent(b.approvalId)}`, { signal: S.abort.signal });
if (res.ok) {
const entry = await res.json();
if (entry && entry.arguments && typeof entry.arguments === 'object') {
@@ -1825,7 +2005,7 @@ async function decide(b, action, withArgs) {
}
let msg;
try {
const res = await fetch(`/api/approvals/${encodeURIComponent(b.approvalId)}/${action}`, {
const res = await fetch(`${S.base}/api/approvals/${encodeURIComponent(b.approvalId)}/${action}`, {
method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify(body), signal: S.abort.signal,
});
if (res.ok) msg = `${action === 'approve' ? 'Approved' : 'Denied'}. Waiting for the run to continue…`;
+56
View File
@@ -466,3 +466,59 @@ def test_empty_state_explains_how_to_record(browser, tmp_path, monkeypatch):
srv.should_exit = True
th.join(timeout=5)
server.reset_state()
def test_embedded_run_box_draws_each_step_live_then_plays_back(browser, tmp_path, monkeypatch):
"""create_replay_app(run=...) mounted in a host app: Run, watch the map, then Play."""
import uvicorn
from starlette.applications import Starlette
from starlette.routing import Mount
from praisonaiui.features.replay import create_replay_app
traces = tmp_path / "traces"
traces.mkdir()
monkeypatch.setenv("PRAISONAI_TRACES_DIR", str(traces))
lines = (FIXTURES / "flow_parallel_loop.jsonl").read_text(encoding="utf-8").splitlines(keepends=True)
seen = []
def runner(text, run_id): # writes a recorded flow slowly, like a live run
seen.append(text)
with open(traces / f"{run_id}.jsonl", "w", encoding="utf-8") as fh:
for line in lines:
fh.write(line)
fh.flush()
time.sleep(0.08)
host = Starlette(routes=[Mount("/replay", app=create_replay_app(runner, input="weekly digest"))])
port = _free_port()
srv = uvicorn.Server(uvicorn.Config(host, host="127.0.0.1", port=port, log_level="warning"))
t = threading.Thread(target=srv.run, daemon=True)
t.start()
for _ in range(100):
if srv.started:
break
time.sleep(0.05)
try:
page, errors = _page(browser)
page.add_init_script("try{localStorage.setItem('rp-view','map')}catch(e){}")
page.goto(f"http://127.0.0.1:{port}/replay/")
page.wait_for_selector(".rp-empty h2:has-text('Nothing has run yet')")
page.get_by_role("button", name="▶ Run").click()
page.wait_for_selector(".rp-node.rp-st-running", timeout=10000) # a step is drawn as running
assert page.locator(".rp-node.rp-unvisited").count() > 0 # later steps not reached yet
page.wait_for_selector(".rp-bar .rp-b-completed", timeout=20000)
assert seen == ["weekly digest"]
assert page.locator(".rp-node.rp-st-running").count() == 0
assert page.get_by_role("button", name="▶ Run").is_enabled()
page.get_by_role("button", name="▶ Play").click() # from the end: restarts at step 1
page.wait_for_timeout(1300)
assert page.locator("[data-key=play]").inner_text() == "❚❚ Pause"
assert page.locator(".rp-node.rp-unvisited").count() > 0 # the map rewound with the cursor
page.locator("[data-key=play]").click()
violations = _axe(page)
assert violations == [], json.dumps(violations, indent=2)
assert not errors, errors
finally:
srv.should_exit = True
t.join(timeout=5)
+59
View File
@@ -397,3 +397,62 @@ test('real-run shape: LLM span encloses the tool call in the same lane', () => {
assert.ok(m.lanes.some((l) => l.id === 'agent:Triage') && m.lanes.some((l) => l.id === 'agent:Billing'));
assert.deepEqual(m.arrows.map((a) => `${a.kind}:${a.from}->${a.to}`), ['handoff:agent:Triage->agent:Billing']);
});
// ── nodeStates: the map at the cursor ─────────────────────────────────
const sev = (seq, event_type, extra = {}) => ({ sequence_num: seq, timestamp: 100 + seq, event_type, ...extra });
test('nodeStates: a step is running between its start and end, then done', () => {
const events = [
sev(0, 'workflow_start'),
sev(1, 'step_start', { node_id: 'step:research' }),
sev(2, 'step_end', { node_id: 'step:research', data: { status: 'completed' } }),
sev(3, 'step_start', { node_id: 'step:draft' }),
];
const nodes = [{ id: 'start', kind: 'start' }, { id: 'step:research' }, { id: 'step:draft' }, { id: 'end', kind: 'end' }];
const at = (i) => Object.fromEntries([...core.nodeStates(events, i, nodes)].map(([k, v]) => [k, v.state]));
assert.deepEqual(at(1), { 'step:research': 'running', start: 'done' });
assert.deepEqual(at(2), { 'step:research': 'done', start: 'done' });
assert.deepEqual(at(3), { 'step:research': 'done', 'step:draft': 'running', start: 'done' });
assert.equal(core.nodeStates(events, -1, nodes).size, 0, 'nothing is reached before the first step');
});
test('nodeStates: end node is reached when the run ends; failed and skipped come from status', () => {
const events = [
sev(0, 'step_start', { node_id: 'step:a' }),
sev(1, 'step_end', { node_id: 'step:a', data: { status: 'failed' } }),
sev(2, 'step_start', { node_id: 'step:b' }),
sev(3, 'step_end', { node_id: 'step:b', data: { status: 'skipped' } }),
sev(4, 'workflow_end'),
];
const st = core.nodeStates(events, Infinity, [{ id: 'end', kind: 'end' }]);
assert.equal(st.get('step:a').state, 'failed');
assert.equal(st.get('step:b').state, 'skipped');
assert.equal(st.get('end').state, 'done');
});
test('nodeStates: runs count only on the node an end event belongs to', () => {
const events = [
sev(0, 'task_start', { node_id: 'task:research', agent_name: 'Researcher' }),
sev(1, 'agent_start', { agent_name: 'Researcher' }),
sev(2, 'agent_end', { agent_name: 'Researcher' }),
sev(3, 'task_end', { node_id: 'task:research', agent_name: 'Researcher', data: { status: 'completed' } }),
sev(4, 'step_start', { node_id: 'step:loop' }),
sev(5, 'step_end', { node_id: 'step:loop' }),
sev(6, 'step_start', { node_id: 'step:loop' }),
sev(7, 'step_end', { node_id: 'step:loop' }),
];
const st = core.nodeStates(events);
assert.equal(st.get('agent:Researcher').runs, 1, 'task_end must not count as another agent run');
assert.equal(st.get('task:research').runs, 1);
assert.equal(st.get('step:loop').runs, 2);
assert.equal(st.get('agent:Researcher').state, 'done');
});
test('nodeStates: a recorded flow fixture ends with every visited node done or failed', () => {
const { events, graph } = load('flow_parallel_loop');
const m = buildModel(events, graph);
const st = core.nodeStates(m.events, Infinity, graph.nodes);
assert.ok(st.size > 0);
for (const [id, v] of st) assert.notEqual(v.state, 'running', `${id} still running after the run ended`);
});
+166
View File
@@ -0,0 +1,166 @@
"""create_replay_app(): the Replay view mounted inside another app, with an optional Run box."""
from __future__ import annotations
import json
import shutil
import threading
import time
from pathlib import Path
import pytest
from starlette.applications import Starlette
from starlette.routing import Mount
from starlette.testclient import TestClient
FIXTURES = Path(__file__).resolve().parents[1] / "fixtures" / "replay"
@pytest.fixture()
def traces(tmp_path, monkeypatch):
d = tmp_path / "traces"
d.mkdir()
monkeypatch.setenv("PRAISONAI_TRACES_DIR", str(d))
monkeypatch.delenv("PRAISON_HOME", raising=False)
return d
def _host(sub) -> TestClient:
"""A host app with the replay app mounted at /replay, like app.mount() in FastAPI."""
return TestClient(Starlette(routes=[Mount("/replay", app=sub)]))
def _cfg(page: str) -> dict:
line = next(l for l in page.splitlines() if l.startswith("render("))
return json.loads(line[line.index(",") + 1 : line.rindex(")")].strip())
def test_page_mounts_the_view_relative_to_the_mount(traces):
from praisonaiui.features.replay import create_replay_app
c = _host(create_replay_app())
r = c.get("/replay/")
assert r.status_code == 200
assert 'from "/replay/replay.js"' in r.text
cfg = _cfg(r.text)
assert cfg == {"base": "/replay", "approve": False} # no run list pin, no Run box
assert "aiui" not in r.text.lower() # no dashboard around it
for name in ("replay.js", "replay-core.js"):
js = c.get(f"/replay/{name}")
assert js.status_code == 200 and js.headers["content-type"].startswith("text/javascript")
def test_page_pins_a_valid_run_and_ignores_a_bad_one(traces):
from praisonaiui.features.replay import create_replay_app
c = _host(create_replay_app())
assert _cfg(c.get("/replay/?run=run-abc").text)["run"] == "run-abc"
page = c.get("/replay/?run=../../etc/passwd").text
assert "run" not in _cfg(page) and "passwd" not in page
def test_api_routes_answer_under_the_mount(traces):
from praisonaiui.features.replay import create_replay_app
shutil.copy(FIXTURES / "team_handoff.jsonl", traces / "team_handoff.jsonl")
c = _host(create_replay_app())
rows = c.get("/replay/api/replay/runs").json()
assert [r["run_id"] for r in rows] == ["team_handoff"]
assert c.get("/replay/api/replay/runs/team_handoff/events").json()["events"]
def test_events_wait_param_answers_a_missing_run_without_404(traces):
from praisonaiui.features.replay import create_replay_app
c = _host(create_replay_app())
assert c.get("/replay/api/replay/runs/run-later/events").status_code == 404
r = c.get("/replay/api/replay/runs/run-later/events?wait=1")
assert r.status_code == 200 and r.json() == {"events": [], "graph": None, "run": None}
def test_no_run_route_without_a_runner(traces):
from praisonaiui.features.replay import create_replay_app
c = _host(create_replay_app())
assert c.post("/replay/api/replay/run", json={"input": "x"}).status_code in (404, 405)
def test_run_box_calls_the_runner_with_a_new_run_id(traces):
from praisonaiui.features.replay import create_replay_app
calls = []
def runner(text, run_id):
(traces / f"{run_id}.jsonl").write_text(json.dumps({"event_type": "session_start", "sequence_num": 0, "timestamp": time.time()}) + "\n")
calls.append((text, run_id))
c = _host(create_replay_app(runner, input="solar sails", placeholder="Topic"))
cfg = _cfg(c.get("/replay/").text)
assert cfg["runner"] == {"input": "solar sails", "placeholder": "Topic", "ask": True}
r = c.post("/replay/api/replay/run", json={"input": "comets"})
assert r.status_code == 200
run_id = r.json()["run_id"]
assert (traces / f"{run_id}.jsonl").is_file(), "answers once the trace exists"
for _ in range(100):
if calls:
break
time.sleep(0.01)
assert calls == [("comets", run_id)]
assert run_id.startswith("run-") and len(run_id) == 16
def test_run_box_without_input_box(traces):
from praisonaiui.features.replay import create_replay_app
cfg = _cfg(_host(create_replay_app(lambda t, r: None, input=None)).get("/replay/").text)
assert cfg["runner"]["ask"] is False
def test_run_rejects_non_json_bad_input_and_a_second_run(traces):
from praisonaiui.features.replay import MAX_RUN_INPUT, create_replay_app
gate = threading.Event()
c = _host(create_replay_app(lambda text, run_id: gate.wait(5)))
assert c.post("/replay/api/replay/run", data="input=x", headers={"content-type": "application/x-www-form-urlencoded"}).status_code == 415
assert c.post("/replay/api/replay/run", content="input=x", headers={"content-type": "text/plain"}).status_code == 415
assert c.post("/replay/api/replay/run", json={"input": 3}).status_code == 400
assert c.post("/replay/api/replay/run", json={"input": "x" * (MAX_RUN_INPUT + 1)}).status_code == 400
assert c.post("/replay/api/replay/run", json={"input": "a"}).status_code == 200
busy = c.post("/replay/api/replay/run", json={"input": "b"})
assert busy.status_code == 409 and "already" in busy.json()["error"]
gate.set()
for _ in range(100): # the lock is released when the runner returns
r = c.post("/replay/api/replay/run", json={"input": "c"})
if r.status_code == 200:
break
time.sleep(0.02)
assert r.status_code == 200
def test_a_failing_runner_frees_the_run_box(traces, caplog):
from praisonaiui.features.replay import create_replay_app
def boom(text, run_id):
raise RuntimeError("no key")
c = _host(create_replay_app(boom))
assert c.post("/replay/api/replay/run", json={"input": ""}).status_code == 200
for _ in range(100):
r = c.post("/replay/api/replay/run", json={"input": ""})
if r.status_code == 200:
break
time.sleep(0.02)
assert r.status_code == 200
def test_async_runner_is_awaited(traces):
from praisonaiui.features.replay import create_replay_app
done = threading.Event()
async def runner(text, run_id):
done.set()
c = _host(create_replay_app(runner))
assert c.post("/replay/api/replay/run", json={"input": ""}).status_code == 200
assert done.wait(3)