feat(testing): add live run dashboard

Author: kvinwang

docs/testing/test-report-output-spec.md

@@ -261,3 +261,14 @@ The output must work offline and inline all CSS, JavaScript, session events, tex
 - collapsible original agent messages, tool calls, command output, and errors;
 - every attachment and the complete raw session; and
 - stable cross-reference anchors.
+
+<a id="report-live-dashboard"></a>
+## 13. Live and historical dashboard
+
+`run-plan --web` starts a read-only HTTP dashboard for case status and the
+native JSONL output of the orchestrator and every case agent. The browser polls
+incremental byte ranges so active output appears without rewriting session
+files. `dstack-test serve --plan <plan> --run-id <run-id>` exposes the same view
+for a completed or interrupted run. The server has no built-in authentication;
+non-loopback binding is permitted only on a trusted network or behind an
+authenticated tunnel.

tools/dstack-test/README.md

@@ -39,6 +39,7 @@ with repeatable `--agent-arg`. The runner writes only below:
 tools/dstack-test/dstack-test run-plan \
   --plan path/to/plan \
   --context run-context.json \
+  --web \
   -- "Follow the environment restrictions in README.md"
 ```
 
@@ -52,6 +53,20 @@ The complete control conversation is stored in `orchestrator.jsonl`.
 
 An interrupted run can be continued with the same run ID and `--resume`.
 
+`--web` starts a read-only live dashboard on `127.0.0.1` and an automatically
+selected port. Use `--web-host 0.0.0.0 --web-port 8000` for remote access;
+there is no built-in authentication, so expose it only on a trusted network or
+through an authenticated tunnel. The dashboard shows case status and polls the
+native orchestrator/case JSONL files for near-real-time output.
+
+After a run has finished, serve the same historical sessions with:
+
+```bash
+tools/dstack-test/dstack-test serve \
+  --plan path/to/plan --run-id run-20260723-001 \
+  --host 127.0.0.1 --port 8000
+```
+
 ## Finalize and validate
 
 ```bash

tools/dstack-test/dstack-test

@@ -26,6 +26,7 @@ from typing import Any
 
 import render
 import tomllib
+import web
 
 TERMINAL_STATUS = ("PASS", "FAIL", "BLOCKED", "NOT_RUN", "SKIPPED")
 
@@ -560,6 +561,59 @@ class FileControlServer:
         self.thread.join()
 
 
+def dashboard_state(plan: render.Plan, run_id: str) -> dict[str, Any]:
+    run_path = plan.root / "results" / run_id / "run.json"
+    run = render.load_json(run_path) if run_path.is_file() else {}
+    counts = {status: 0 for status in (*TERMINAL_STATUS, "RUNNING", "PENDING")}
+    cases = []
+    for case in plan.cases:
+        result_path = case.path / "results" / run_id / "result.json"
+        session_path = case.path / "results" / run_id / "session.jsonl"
+        if result_path.is_file():
+            status = str(render.load_json(result_path).get("status", "PENDING"))
+        else:
+            status = "RUNNING" if session_path.is_file() else "PENDING"
+        counts[status] = counts.get(status, 0) + 1
+        cases.append({"id": case.id, "title": case.title, "status": status})
+    orchestrator = plan.root / "results" / run_id / "orchestrator.jsonl"
+    return {
+        "title": plan.index.get("title", plan.index["id"]),
+        "run_id": run_id,
+        "run_status": run.get("status", "RUNNING"),
+        "orchestrator_status": "PASS"
+        if run_path.is_file()
+        else ("RUNNING" if orchestrator.is_file() else "PENDING"),
+        "counts": counts,
+        "cases": cases,
+    }
+
+
+def dashboard_log(
+    plan: render.Plan, run_id: str, agent: str, offset: int
+) -> dict[str, Any]:
+    if agent == "orchestrator":
+        path = plan.root / "results" / run_id / "orchestrator.jsonl"
+    elif agent.startswith("case:"):
+        case = find_case(plan, agent.removeprefix("case:"))
+        path = case.path / "results" / run_id / "session.jsonl"
+    else:
+        raise DstackTestError("unknown agent session")
+    if not path.is_file():
+        return {"text": "", "next_offset": 0, "reset": offset > 0}
+    size = path.stat().st_size
+    reset = offset > size
+    with path.open("rb") as source:
+        source.seek(0 if reset else offset)
+        data = source.read(1024 * 1024)
+        next_offset = source.tell()
+    return {
+        "text": data.decode("utf-8", errors="replace"),
+        "next_offset": next_offset,
+        "reset": reset,
+        "eof": next_offset >= size,
+    }
+
+
 def write_sha256s(root: Path) -> None:
     entries: list[str] = []
     for path in sorted(
@@ -806,6 +860,9 @@ def build_parser() -> argparse.ArgumentParser:
     run_plan.add_argument("--agent-arg", action="append", default=[])
     run_plan.add_argument("--overwrite", action="store_true")
     run_plan.add_argument("--resume", action="store_true")
+    run_plan.add_argument("--web", action="store_true", help="serve a live dashboard")
+    run_plan.add_argument("--web-host", default="127.0.0.1")
+    run_plan.add_argument("--web-port", type=int, default=0)
     run_plan.add_argument("prompt", nargs=argparse.REMAINDER)
 
     sub.add_parser("next-case", help="return the next unresolved controlled case")
@@ -835,6 +892,11 @@ def build_parser() -> argparse.ArgumentParser:
     )
     add_common_plan(render_parser)
     render_parser.add_argument("--output", type=Path, required=True)
+
+    serve = sub.add_parser("serve", help="serve a live or historical run dashboard")
+    add_common_plan(serve)
+    serve.add_argument("--host", default="127.0.0.1")
+    serve.add_argument("--port", type=int, default=8000)
     return parser
 
 
@@ -908,6 +970,21 @@ def main(argv: list[str] | None = None) -> int:
         control_dir = run_dir / "control"
         server = FileControlServer(state, control_dir)
         server.start()
+        dashboard = None
+        if args.web:
+            dashboard = web.Dashboard(
+                lambda: dashboard_state(plan, args.run_id),
+                lambda agent, offset: dashboard_log(plan, args.run_id, agent, offset),
+                args.web_host,
+                args.web_port,
+            )
+            dashboard.start()
+            web_host, web_port = dashboard.address
+            print(
+                f"dashboard: http://{web_host}:{web_port}/",
+                file=sys.stderr,
+                flush=True,
+            )
         prompt = orchestration_prompt(plan, prompt_text(args))
         (run_dir / "orchestrator-prompt.md").write_text(prompt, encoding="utf-8")
         environment = os.environ.copy()
@@ -931,6 +1008,8 @@ def main(argv: list[str] | None = None) -> int:
         finally:
             server.close()
             shutil.rmtree(control_dir, ignore_errors=True)
+            if dashboard:
+                dashboard.close()
         events, detected_model = validate_session(run_dir / "orchestrator.jsonl")
         atomic_json(
             run_dir / "orchestrator-runner.json",
@@ -968,6 +1047,21 @@ def main(argv: list[str] | None = None) -> int:
             "output": str(args.output),
             "bytes": args.output.stat().st_size,
         }
+    elif args.subcommand == "serve":
+        dashboard = web.Dashboard(
+            lambda: dashboard_state(plan, args.run_id),
+            lambda agent, offset: dashboard_log(plan, args.run_id, agent, offset),
+            args.host,
+            args.port,
+        )
+        dashboard.start()
+        host, port = dashboard.address
+        print(f"dashboard: http://{host}:{port}/", file=sys.stderr, flush=True)
+        try:
+            dashboard.thread.join()
+        except KeyboardInterrupt:
+            dashboard.close()
+        return 0
     else:
         raise AssertionError(args.subcommand)
     print(json.dumps(value, ensure_ascii=False, indent=2))

tools/dstack-test/tests/test_dstack_test.py

@@ -81,6 +81,14 @@ def test_validate_and_render_fixture(self) -> None:
         self.assertIn("198.51.100.27:45678", output)
         self.assertIn("session-tc-gw-pp-001-event-2", output)
 
+    def test_dashboard_exposes_historical_status_and_log(self) -> None:
+        plan = render.load_plan(FIXTURE)
+        state = dstack_test.dashboard_state(plan, "run-demo")
+        self.assertEqual(state["cases"][0]["status"], "PASS")
+        log = dstack_test.dashboard_log(plan, "run-demo", "case:tc-gw-pp-001", 0)
+        self.assertGreater(log["next_offset"], 0)
+        self.assertIn("thread.started", log["text"])
+
     def test_finalize_rebuilds_run_summary(self) -> None:
         with tempfile.TemporaryDirectory() as temporary:
             plan_path = self.copy_fixture(Path(temporary))

tools/dstack-test/web.py

@@ -0,0 +1,82 @@
+# SPDX-FileCopyrightText: © 2026 Phala Network <dstack@phala.network>
+# SPDX-License-Identifier: Apache-2.0
+"""Embedded live dashboard for dstack test runs."""
+
+import http.server
+import json
+import threading
+import urllib.parse
+from typing import Any, Callable
+
+HTML = r"""<!doctype html><html><head><meta charset="utf-8"><meta name="viewport" content="width=device-width"><title>dstack test</title><style>
+:root{color-scheme:dark;font:14px system-ui;background:#0b1020;color:#dce5ff}body{margin:0}header{padding:16px 22px;background:#121a30}h1{margin:0;font-size:20px}.meta{color:#91a4cc}.grid{display:grid;grid-template-columns:360px 1fr;min-height:calc(100vh - 65px)}aside,main{padding:16px}aside{border-right:1px solid #273453}.case{display:block;width:100%;text-align:left;background:#151e35;color:inherit;border:1px solid #293858;border-radius:8px;padding:10px;margin:7px 0;cursor:pointer}.case:hover,.active{border-color:#6da7ff}.badge{float:right;border-radius:10px;padding:2px 7px;font-size:11px}.PASS{background:#126b47}.FAIL{background:#8b2733}.SKIPPED,.BLOCKED{background:#74551b}.RUNNING{background:#215a91}.PENDING{background:#39445d}.summary{display:flex;gap:6px;flex-wrap:wrap}.summary span{background:#1a2540;padding:5px 8px;border-radius:12px}button{color:inherit}pre{white-space:pre-wrap;word-break:break-word;background:#070b14;border:1px solid #273453;border-radius:8px;padding:14px;height:calc(100vh - 135px);overflow:auto}.toolbar{display:flex;gap:10px;align-items:center}@media(max-width:800px){.grid{grid-template-columns:1fr}}</style></head><body><header><h1 id="title">dstack test</h1><div class="meta" id="meta">Connecting…</div></header><div class="grid"><aside><div class="summary" id="summary"></div><div id="cases"></div></aside><main><div class="toolbar"><strong id="agent">Plan orchestrator</strong><button onclick="resetLog()">Reload history</button><label><input id="follow" type="checkbox" checked> follow</label></div><pre id="log"></pre></main></div><script>
+let selected='orchestrator',offset=0;const log=document.querySelector('#log');const esc=s=>String(s).replace(/[&<>"']/g,c=>({'&':'&amp;','<':'&lt;','>':'&gt;','"':'&quot;',"'":'&#39;'}[c]));
+function pick(id,label){selected=id;offset=0;log.textContent='';document.querySelector('#agent').textContent=label;pollLog()}function resetLog(){offset=0;log.textContent='';pollLog()}
+async function pollState(){try{const s=await(await fetch('/api/state')).json();title.textContent=s.title;meta.textContent=`run ${s.run_id} · ${s.run_status}`;summary.innerHTML=Object.entries(s.counts).filter(x=>x[1]).map(x=>`<span>${esc(x[0])} ${x[1]}</span>`).join('');let h=`<button class="case" onclick="pick('orchestrator','Plan orchestrator')">Plan orchestrator<span class="badge ${s.orchestrator_status}">${s.orchestrator_status}</span></button>`;for(const c of s.cases)h+=`<button class="case" onclick="pick('case:${esc(c.id)}','${esc(c.id)}')">${esc(c.id)}<span class="badge ${c.status}">${c.status}</span><br><small>${esc(c.title)}</small></button>`;cases.innerHTML=h}catch(e){meta.textContent=e}}
+async function pollLog(){try{const r=await(await fetch(`/api/log?agent=${encodeURIComponent(selected)}&offset=${offset}`)).json();if(r.reset){offset=0;log.textContent=''}if(r.text){log.textContent+=r.text;offset=r.next_offset;if(follow.checked)log.scrollTop=log.scrollHeight}}catch(e){}}
+setInterval(pollState,1000);setInterval(pollLog,500);pollState();pollLog();</script></body></html>"""
+
+
+class Dashboard:
+    """Serve a read-only live run dashboard."""
+
+    def __init__(
+        self,
+        state: Callable[[], dict[str, Any]],
+        log: Callable[[str, int], dict[str, Any]],
+        host: str,
+        port: int,
+    ):
+        class Handler(http.server.BaseHTTPRequestHandler):
+            def log_message(self, _format: str, *_args: Any) -> None:
+                pass
+
+            def reply(self, value: Any, status: int = 200) -> None:
+                data = json.dumps(value, ensure_ascii=False).encode()
+                self.send_response(status)
+                self.send_header("Content-Type", "application/json; charset=utf-8")
+                self.send_header("Cache-Control", "no-store")
+                self.send_header("Content-Length", str(len(data)))
+                self.end_headers()
+                self.wfile.write(data)
+
+            def do_GET(self) -> None:
+                parsed = urllib.parse.urlparse(self.path)
+                if parsed.path == "/":
+                    data = HTML.encode()
+                    self.send_response(200)
+                    self.send_header("Content-Type", "text/html; charset=utf-8")
+                    self.send_header("Content-Length", str(len(data)))
+                    self.end_headers()
+                    self.wfile.write(data)
+                elif parsed.path == "/api/state":
+                    self.reply(state())
+                elif parsed.path == "/api/log":
+                    query = urllib.parse.parse_qs(parsed.query)
+                    try:
+                        self.reply(
+                            log(
+                                query.get("agent", [""])[0],
+                                int(query.get("offset", ["0"])[0]),
+                            )
+                        )
+                    except Exception as error:  # noqa: BLE001 - API boundary
+                        self.reply({"error": str(error)}, 400)
+                else:
+                    self.send_error(404)
+
+        self.server = http.server.ThreadingHTTPServer((host, port), Handler)
+        self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
+
+    @property
+    def address(self) -> tuple[str, int]:
+        host, port = self.server.server_address[:2]
+        return str(host), int(port)
+
+    def start(self) -> None:
+        self.thread.start()
+
+    def close(self) -> None:
+        self.server.shutdown()
+        self.server.server_close()
+        self.thread.join()