approaches: strategies tried →
@@ -865,6 +894,15 @@ def run_page(label: str, idx: int):
raw.append((s.iteration, s.score))
raw_std.append(s.metrics.get("combined_score_std") if s.metrics else None)
has_std = any(isinstance(x, (int, float)) and x > 0 for x in raw_std)
+ # Live feed: if the run wrote a per-iteration progress.jsonl, draw the
+ # trajectory from it (finer than checkpoints). Falls back to checkpoint
+ # steps for older runs.
+ progress = explore.read_progress(rd)
+ has_progress = bool(progress)
+ if has_progress:
+ pbest, praw = explore.progress_trajectory(progress)
+ if pbest:
+ best, raw, raw_std = pbest, praw, [None] * len(praw)
chart = (
svg.trajectory(best, raw, raw_std=raw_std)
if best
@@ -915,11 +953,53 @@ def run_page(label: str, idx: int):
calls=rs.calls,
chart=chart,
has_std=has_std,
+ has_progress=has_progress,
+ is_active=rs.is_active,
has_solution=solution.find_visualizer(rd, roots) is not None,
models=models,
step_rows=step_rows,
)
+ # ── live JSON for the dashboard poller: current chart + headline stats ──────
+ @app.get("/setup/{label}/run/{idx}/live.json")
+ def run_live(label: str, idx: int):
+ from fastapi.responses import JSONResponse
+
+ arm = arm_or_404(label)
+ if idx < 0 or idx >= len(arm.run_dirs):
+ raise HTTPException(404, f"run {idx} out of range for {label}")
+ rd = arm.run_dirs[idx]
+ rs = explore.run_state(rd)
+ progress = explore.read_progress(rd)
+ if progress:
+ best, raw = explore.progress_trajectory(progress)
+ iteration = max((r.get("iter", 0) for r in progress), default=0)
+ best_score = max(
+ (r.get("best") for r in progress if isinstance(r.get("best"), (int, float))),
+ default=None,
+ )
+ else:
+ steps = explore.steps_for_run(rd)
+ best, raw, b = [], [], float("-inf")
+ for s in steps:
+ if s.score is not None:
+ b = max(b, s.score)
+ best.append((s.iteration, b))
+ raw.append((s.iteration, s.score))
+ iteration = len(steps)
+ best_score = rs.best_score
+ chart = svg.trajectory(best, raw) if best else ""
+ return JSONResponse(
+ {
+ "active": rs.is_active,
+ "iteration": iteration,
+ "best": best_score,
+ "cost": _fmt_cost(rs.cost),
+ "calls": rs.calls,
+ "chart": chart,
+ }
+ )
+
# ── drawer fragment for one step (notes + diff + optional viz iframe) ───────
@app.get("/setup/{label}/run/{idx}/step/{iteration}/detail", response_class=HTMLResponse)
def step_detail(label: str, idx: int, iteration: int):
diff --git a/tests/test_live.py b/tests/test_live.py
new file mode 100644
index 0000000..60ed854
--- /dev/null
+++ b/tests/test_live.py
@@ -0,0 +1,116 @@
+"""Tests for live-run streaming: the per-iteration progress feed (progress.jsonl),
+its readers in kaievolve.explore, and the viewer's live badge + live.json endpoint.
+"""
+
+import json
+import time
+import unittest
+from pathlib import Path
+from tempfile import TemporaryDirectory
+
+from kaievolve import explore
+
+
+def _write_progress(run_dir: Path, rows):
+ run_dir.mkdir(parents=True, exist_ok=True)
+ with open(run_dir / "progress.jsonl", "w") as f:
+ for r in rows:
+ f.write(json.dumps(r) + "\n")
+
+
+_ROWS = [
+ {"iter": 0, "score": 0.30, "best": 0.30, "program_id": "p0", "model": "m", "ts": 1.0},
+ {"iter": 1, "score": 0.20, "best": 0.30, "program_id": "p1", "model": "m", "ts": 2.0},
+ {"iter": 2, "score": 0.55, "best": 0.55, "program_id": "p2", "model": "m", "ts": 3.0},
+ {"iter": 3, "score": 0.62, "best": 0.62, "program_id": "p3", "model": "m", "ts": 4.0},
+]
+
+
+class TestProgressFeed(unittest.TestCase):
+ def test_read_progress_absent_is_empty(self):
+ with TemporaryDirectory() as d:
+ self.assertEqual(explore.read_progress(Path(d)), [])
+
+ def test_progress_trajectory(self):
+ best, raw = explore.progress_trajectory(_ROWS)
+ # best is the best-so-far step line; raw is every attempt
+ self.assertEqual(best, [(0, 0.30), (1, 0.30), (2, 0.55), (3, 0.62)])
+ self.assertEqual(len(raw), 4)
+ self.assertEqual(raw[1], (1, 0.20))
+
+ def test_run_state_live_override(self):
+ with TemporaryDirectory() as d:
+ rd = Path(d) / "setup" / "task" / "run_3"
+ _write_progress(rd, _ROWS)
+ # a recent usage_log so the run reads as active
+ (rd / "usage_log.jsonl").write_text(
+ json.dumps({"model": "m", "cost": 0.01, "ts": time.time()}) + "\n"
+ )
+ st = explore.run_state(rd)
+ self.assertEqual(st.iteration, 3) # current iteration from the feed
+ self.assertAlmostEqual(st.best_score, 0.62)
+ self.assertEqual(st.curve, [0.30, 0.30, 0.55, 0.62]) # finer than checkpoints
+ self.assertTrue(st.is_active) # recent activity
+
+
+def _has_viewer_deps() -> bool:
+ try:
+ import fastapi # noqa: F401
+ import jinja2 # noqa: F401
+ from fastapi.testclient import TestClient # noqa: F401
+
+ return True
+ except ImportError:
+ return False
+
+
+@unittest.skipUnless(_has_viewer_deps(), "viewer extra not installed")
+class TestViewerLive(unittest.TestCase):
+ def _client(self, active: bool):
+ from fastapi.testclient import TestClient
+ from kaievolve.viewer.server import create_app
+
+ self._tmp = TemporaryDirectory()
+ root = Path(self._tmp.name) / "bench"
+ rd = root / "auto_full" / "task" / "run_9"
+ _write_progress(rd, _ROWS)
+ (rd / "usage_log.jsonl").write_text(
+ json.dumps({"model": "m", "cost": 0.02, "total_tokens": 9, "ts": time.time()}) + "\n"
+ )
+ (root / "auto_full" / "results.jsonl").write_text(json.dumps({"best_score": 0.62}) + "\n")
+ if not active:
+ # idle = the run's files haven't been touched in a while. is_active is
+ # mtime-based (a freshly-appended progress feed reads as live), so age
+ # the files well past the 300s activity window.
+ import os
+
+ old = time.time() - 3600
+ for p in (rd / "progress.jsonl", rd / "usage_log.jsonl"):
+ os.utime(p, (old, old))
+ return TestClient(create_app(root, None))
+
+ def test_dashboard_live_badge_and_endpoint(self):
+ c = self._client(active=True)
+ page = c.get("/setup/auto_full/run/0")
+ self.assertEqual(page.status_code, 200)
+ # live badge shown (not hidden) + the streaming note
+ self.assertIn('id="livedot"', page.text)
+ self.assertNotIn('id="livedot" hidden', page.text)
+ self.assertIn("streaming live", page.text)
+ lj = c.get("/setup/auto_full/run/0/live.json").json()
+ self.assertTrue(lj["active"])
+ self.assertEqual(lj["iteration"], 3)
+ self.assertAlmostEqual(lj["best"], 0.62)
+ self.assertIn("