feat(dashboard): serve run inventory from Redis at /api/v1/inspect/inventory

Makefile

@@ -19,7 +19,7 @@ HOST           ?= 127.0.0.1
 PORT           ?= 9003
 # The fully qualified base URL used by local CLI tools and clients
 BASE_URL       ?= http://$(HOST):$(PORT)
-UNIT_TESTS ?= tests.test_session_lifecycle tests.test_proto_codec tests.test_forward_only tests.test_fft_batch_failure tests.test_api_server_paths tests.test_accel_timeslicer tests.test_trainer_optimizer_correctness tests.test_trainer_logprob_chunks tests.test_worker_manager tests.test_scheduler_worker_manager tests.test_estimator tests.test_redis_store tests.test_cluster_eval_script tests.test_perf_compare_script tests.test_delta_weight_sync tests.test_diffing_backends tests.test_sampler_weight_rotation tests.test_sampler_batch_limits tests.test_commands tests.test_weight_sync_config tests.test_exclusive_trainer tests.test_delete_model tests.test_automodel_worker tests.test_worker_thread_device tests.test_trainer_data_parallel tests.test_openrl_status tests.test_skill_docs tests.test_trainer_ready tests.test_sampling_store tests.test_lora_snapshots tests.test_pool_accept tests.test_sampling_dispatcher tests.test_vllm_sampler tests.test_vllm_options
+UNIT_TESTS ?= tests.test_session_lifecycle tests.test_proto_codec tests.test_forward_only tests.test_fft_batch_failure tests.test_api_server_paths tests.test_accel_timeslicer tests.test_trainer_optimizer_correctness tests.test_trainer_logprob_chunks tests.test_worker_manager tests.test_scheduler_worker_manager tests.test_estimator tests.test_redis_store tests.test_cluster_eval_script tests.test_perf_compare_script tests.test_delta_weight_sync tests.test_diffing_backends tests.test_sampler_weight_rotation tests.test_sampler_batch_limits tests.test_commands tests.test_weight_sync_config tests.test_exclusive_trainer tests.test_delete_model tests.test_automodel_worker tests.test_worker_thread_device tests.test_trainer_data_parallel tests.test_openrl_status tests.test_skill_docs tests.test_trainer_ready tests.test_sampling_store tests.test_lora_snapshots tests.test_pool_accept tests.test_sampling_dispatcher tests.test_vllm_sampler tests.test_vllm_options tests.test_dashboard
 # Only forward BASE_URL to e2e when the user supplied it. The Makefile default
 # is for local CLI usage; e2e should start its own backend by default.
 TRAINING_TEST_BASE_URL ?= $(if $(filter environment command line,$(origin BASE_URL)),$(BASE_URL),)

docs/configuration.md

@@ -122,3 +122,13 @@ for. On one machine every pool has one replica, in process. See
 | `OPEN_RL_DISPATCH_CONCURRENCY` | `256` | Requests one pool keeps in flight. |
 | `OPEN_RL_SAMPLE_DEADLINE_SECONDS` | `3600` | How long a sampling request waits for a result before it fails. Set it on the API server. Worker pods and pool dispatchers get the same value. |
 | `OPEN_RL_DISPATCHER_IMAGE` | the worker image | Image for each pool's dispatcher container. The release sets it to the API-server image, which carries the dispatcher without the GPU stack and pulls in seconds. |
+
+## Inspection API
+
+The API server serves a read-only inventory of runs at
+`/api/v1/inspect/inventory`. It reads runs and their live sessions from the
+same Redis as the rest of the server.
+
+| Env var | Default | What it does |
+| --- | --- | --- |
+| `REDIS_URL` | unset | The Redis that holds model records and live sessions. Unset, the inventory has no runs. |

src/server/api_server.py

@@ -27,6 +27,7 @@
 from starlette.exceptions import HTTPException as StarletteHTTPException
 
 from server import openrl_status, proto_codec
+from server.dashboard.router import router
 from server.lora_snapshots import snapshot_name, snapshot_root
 from server.model_metadata import (
   TrainingModelMetadata,
@@ -809,6 +810,7 @@ def openapi_with_settings() -> dict[str, Any]:
 
 app.openapi = openapi_with_settings
 FastAPIInstrumentor.instrument_app(app, excluded_urls="/api/v1/retrieve_future,/api/v1/session_heartbeat")
+app.include_router(router)
 
 
 @app.exception_handler(RequestValidationError)

src/server/dashboard/fleet.py

@@ -0,0 +1,80 @@
+import asyncio
+import json
+import time
+from dataclasses import dataclass
+from datetime import UTC, datetime
+from typing import Literal
+
+from server import openrl_status
+from server.model_metadata import decode_model_metadata
+from server.store import get_state_store
+
+CACHE_SECONDS, STALE_SECONDS, READ_TIMEOUT_SECONDS = 5, 60, 6
+Lifecycle = Literal["ended", "active"]
+
+
+@dataclass
+class Entry:
+  at: float = 0.0
+  task: asyncio.Future | None = None
+  last_good: asyncio.Future | None = None
+  last_good_at: float = 0.0
+
+
+cache: dict[str, Entry] = {}
+
+
+def succeeded(task: asyncio.Future | None) -> bool:
+  return task is not None and task.done() and not task.cancelled() and task.exception() is None
+
+
+async def cached(name: str, seconds: float, read) -> object:
+  entry = cache.setdefault(name, Entry())
+  task = entry.task
+  if succeeded(task):
+    entry.last_good, entry.last_good_at = task, entry.at
+  if task is None or (task.done() and (not succeeded(task) or time.monotonic() - entry.at > seconds)):
+    entry.at, entry.task = time.monotonic(), asyncio.ensure_future(read())
+    entry.task.add_done_callback(lambda t: t.cancelled() or t.exception())  # retrieved, so asyncio does not log it
+  if succeeded(entry.task):
+    return entry.task.result()
+  if entry.last_good is not None and time.monotonic() - entry.last_good_at <= STALE_SECONDS:
+    return entry.last_good.result()
+  return await asyncio.shield(entry.task)
+
+
+async def snapshot() -> dict:
+  return await cached("inventory", CACHE_SECONDS, read)
+
+
+def lifecycle(status: str, live: bool) -> Lifecycle:
+  return "active" if status == "active" and live else "ended"
+
+
+def join(run_id: str, raw: str, live: set[str]) -> dict:
+  row, meta = json.loads(raw), decode_model_metadata(raw)
+  name = row.get("name") or row.get("run_name")
+  run = {"run_id": run_id, "name": name or f"{(meta.base_model or 'Run').split('/')[-1]} · {run_id[:8]}", "display_name": name}
+  run.update(model=meta.base_model, recipe_name=row.get("recipe_name"), fine_tuning_type=meta.fine_tuning_type, client_status=meta.status)
+  run["lifecycle"] = lifecycle(meta.status, any(s in live for s in meta.session_ids))
+  run.update(created_at=meta.created_at, completed_at=meta.completed_at, session_ids=meta.session_ids, settings=openrl_status.settings_of(meta))
+  return run | {"shared_runtime": meta.runtime(run_id) != run_id}
+
+
+async def read() -> dict:
+  store, observed_at = get_state_store(), datetime.now(UTC).isoformat(timespec="seconds")
+  scans = [asyncio.wait_for(store.scan_values(prefix), READ_TIMEOUT_SECONDS) for prefix in ("open_rl:model_meta:", "open_rl:session:")]
+  metas, sessions = await asyncio.gather(*scans, return_exceptions=True)
+  errors, runs = [], []
+  if failed := next((r for r in (metas, sessions) if isinstance(r, Exception)), None):
+    errors.append(f"Redis read failed: {f'no reply in {READ_TIMEOUT_SECONDS} s' if isinstance(failed, TimeoutError) else failed}")
+    metas, sessions = {}, {}
+  live = {name.removeprefix("open_rl:session:") for name in sessions}
+  for name, raw in metas.items():
+    run_id = name.removeprefix("open_rl:model_meta:")
+    try:
+      runs.append(join(run_id, raw, live))
+    except Exception as exc:
+      errors.append(f"model record {run_id} unreadable: {exc}")
+  runs.sort(key=lambda r: -r["created_at"])
+  return {"observed_at": observed_at, "errors": errors, "runs": runs}

src/server/dashboard/router.py

@@ -0,0 +1,13 @@
+from fastapi import APIRouter
+
+from server.dashboard import fleet
+
+router = APIRouter(tags=["inspection"])
+
+
+@router.get("/api/v1/inspect/inventory")
+async def inventory():
+  """Runs with their lifecycle (ended, active) and an error per unavailable source or unreadable run record. client_status is
+  what the client last reported, active until it deletes the model, so a crashed run stays active there; lifecycle is ended
+  once none of the run's sessions is live."""
+  return await fleet.snapshot()

src/server/store.py

@@ -269,6 +269,11 @@ async def remove_from_set(self, key: str, member: str) -> None:
   async def set_members(self, key: str) -> set[str]:
     pass
 
+  @abstractmethod
+  async def scan_values(self, prefix: str) -> dict[str, str]:
+    """Every live value whose key starts with prefix."""
+    pass
+
 
 class InMemoryStateStore(StateStore):
   def __init__(self):
@@ -306,6 +311,9 @@ async def remove_from_set(self, key: str, member: str) -> None:
   async def set_members(self, key: str) -> set[str]:
     return set(self.sets.get(key, set()))
 
+  async def scan_values(self, prefix: str) -> dict[str, str]:
+    return {key: value for key in list(self.kv_store) if key.startswith(prefix) and (value := self.get_value_sync(key)) is not None}
+
 
 class RedisStateStore(StateStore):
   def __init__(self, client: redis.Redis, sync_client: sync_redis.Redis):
@@ -334,6 +342,10 @@ async def remove_from_set(self, key: str, member: str) -> None:
   async def set_members(self, key: str) -> set[str]:
     return set(await self.redis.smembers(key))
 
+  async def scan_values(self, prefix: str) -> dict[str, str]:
+    keys = [key async for key in self.redis.scan_iter(match=prefix + "*", count=1000)]
+    return {key: value for key, value in zip(keys, await self.redis.mget(keys)) if value is not None} if keys else {}
+
 
 @cache
 def redis_client(redis_url: str) -> redis.Redis:

tests/test_dashboard.py

@@ -0,0 +1,48 @@
+import asyncio
+import unittest
+from unittest import mock
+
+from fastapi.testclient import TestClient
+
+from server import api_server
+from server.dashboard import fleet
+from server.model_metadata import TrainingModelMetadata
+from server.store import InMemoryStateStore
+
+
+def meta(**fields) -> str:
+  return TrainingModelMetadata(base_model="Qwen/Qwen3-8B", **fields).model_dump_json()
+
+
+class InventoryTest(unittest.TestCase):
+  def inventory(self, **records: str) -> dict:
+    store = InMemoryStateStore()
+    for key, value in records.items():
+      asyncio.run(store.set_value(f"open_rl:{key}", value))
+    fleet.cache.clear()
+    self.enterContext(mock.patch.object(fleet, "get_state_store", return_value=store))
+    return TestClient(api_server.app).get("/api/v1/inspect/inventory").json()
+
+  def test_a_run_is_active_only_while_the_client_has_a_live_session(self):
+    body = self.inventory(
+      **{
+        "model_meta:live": meta(session_ids=["s1"], created_at=3),
+        "model_meta:idle": meta(session_ids=["s2"], created_at=2),
+        "model_meta:done": meta(status="completed", session_ids=["s1"], created_at=1),
+        "session:s1": "{}",
+      }
+    )
+
+    runs = [(r["run_id"], r["lifecycle"], r["client_status"]) for r in body["runs"]]
+    self.assertEqual(runs, [("live", "active", "active"), ("idle", "ended", "active"), ("done", "ended", "completed")])
+    self.assertEqual(body["errors"], [])
+
+  def test_a_malformed_model_record_is_skipped_with_an_error(self):
+    body = self.inventory(**{"model_meta:good": meta(), "model_meta:bad": "not json"})
+
+    self.assertEqual([r["run_id"] for r in body["runs"]], ["good"])
+    self.assertTrue(any("bad" in error for error in body["errors"]), body["errors"])
+
+
+if __name__ == "__main__":
+  unittest.main()

tests/test_redis_store.py

@@ -151,6 +151,15 @@ async def test_values_expire_and_sets_hold_members(self) -> None:
     self.assertEqual(await self.state.set_members("s"), {"b"})
     self.assertEqual(await self.state.set_members("missing"), set())
 
+  async def test_scan_values_returns_live_values_under_a_prefix(self) -> None:
+    await self.state.set_value("open_rl:model_meta:a", "1")
+    await self.state.set_value("open_rl:model_meta:b", "2")
+    await self.state.set_value("open_rl:model_meta:gone", "3", ttl_seconds=0.2)
+    await self.state.set_value("open_rl:session:a", "4")
+    await asyncio.sleep(0.5)
+    self.assertEqual(await self.state.scan_values("open_rl:model_meta:"), {"open_rl:model_meta:a": "1", "open_rl:model_meta:b": "2"})
+    self.assertEqual(await self.state.scan_values("open_rl:missing:"), {})
+
 
 class InMemoryStoreTest(unittest.IsolatedAsyncioTestCase):
   def setUp(self) -> None:
@@ -203,6 +212,15 @@ async def test_values_expire_and_sets_hold_members(self) -> None:
     self.assertEqual(await self.state.set_members("s"), {"b"})
     self.assertEqual(await self.state.set_members("missing"), set())
 
+  async def test_scan_values_returns_live_values_under_a_prefix(self) -> None:
+    await self.state.set_value("open_rl:model_meta:a", "1")
+    await self.state.set_value("open_rl:model_meta:b", "2")
+    await self.state.set_value("open_rl:model_meta:gone", "3", ttl_seconds=0.2)
+    await self.state.set_value("open_rl:session:a", "4")
+    await asyncio.sleep(0.5)
+    self.assertEqual(await self.state.scan_values("open_rl:model_meta:"), {"open_rl:model_meta:a": "1", "open_rl:model_meta:b": "2"})
+    self.assertEqual(await self.state.scan_values("open_rl:missing:"), {})
+
 
 if __name__ == "__main__":
   unittest.main()