feat(server): add runtime systems status registry

docs/runtime-systems.md

@@ -0,0 +1,11 @@
+# Read-only runtime systems
+
+`GET /v1/mtplx/systems` uses the same authentication and rate-limit middleware as other `/v1` endpoints. The server now publishes a concrete `serving` provider from its existing CPU-side state whenever this endpoint is read. No separate polling thread or second runtime controller is introduced.
+
+The provider exposes loaded/released availability, AR/MTP mode, foreground/dashboard request counts, completed/cancelled counters, and whether MTP is enabled. Its phase changes between unavailable, unknown, idle and busy; unknown counters are null, not invented zeroes. These aggregate scalars are sampled sequentially, not promised to be a transactional multi-counter snapshot.
+
+Only fixed enums, booleans and bounded integers are published. Model paths, API keys, request/client IDs, prompt/tool content and exception messages are not copied. A failed refresh replaces prior healthy data with an unavailable status and fixed reason rather than leaving stale success visible. Publication is serialized per provider and the registry returns detached JSON.
+
+The provider never loads or runs a model, acquires the model lock, resizes caches, changes configuration or touches the Metal allocator. Existing inference/runtime owners remain authoritative. The generic registry does not import this adapter; the server wires it as an optional refresh callback. Other components can publish independent bounded statuses through the same registry.
+
+The Systems dashboard in #365 can render this actual provider without a placeholder. Production-app tests use the real `create_app` and HTTP route with an injected runtime test state, including authentication and live counter changes. That proves the wiring and contract, not a physical-model or packaged-desktop smoke.

mtplx/runtime_systems.py

@@ -0,0 +1,153 @@
+"""Provider-neutral runtime system status registry.
+
+The registry is an observability boundary. Runtime components may publish
+JSON-compatible status without the registry importing or controlling them.
+"""
+
+from __future__ import annotations
+
+import json
+import re
+import threading
+import time
+from collections.abc import Callable, Mapping
+from typing import Any
+
+_SYSTEM_NAME = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:-]{0,127}$")
+
+
+class RuntimeSystemsRegistry:
+    """Thread-safe, bounded collection of runtime status snapshots."""
+
+    def __init__(
+        self,
+        *,
+        max_systems: int = 128,
+        max_status_bytes: int = 64 * 1024,
+        clock: Callable[[], float] = time.time,
+    ) -> None:
+        if max_systems < 1:
+            raise ValueError("max_systems must be positive")
+        if max_status_bytes < 2:
+            raise ValueError("max_status_bytes must be at least 2")
+        self._max_systems = int(max_systems)
+        self._max_status_bytes = int(max_status_bytes)
+        self._clock = clock
+        self._lock = threading.Lock()
+        self._revision = 0
+        self._updated_at_s = float(clock())
+        self._systems: dict[str, dict[str, Any]] = {}
+
+    @staticmethod
+    def _validate_name(name: str) -> str:
+        normalized = str(name).strip()
+        if not _SYSTEM_NAME.fullmatch(normalized):
+            raise ValueError(
+                "system name must be 1 to 128 characters and contain only "
+                "letters, digits, dot, colon, underscore, or hyphen"
+            )
+        return normalized
+
+    def _clone_status(self, status: Mapping[str, Any]) -> dict[str, Any]:
+        if not isinstance(status, Mapping):
+            raise TypeError("status must be a mapping")
+        try:
+            encoded = json.dumps(
+                dict(status),
+                allow_nan=False,
+                ensure_ascii=False,
+                separators=(",", ":"),
+            ).encode("utf-8")
+        except (TypeError, ValueError) as exc:
+            raise ValueError("status must be JSON-compatible") from exc
+        if len(encoded) > self._max_status_bytes:
+            raise ValueError(
+                f"status exceeds max_status_bytes={self._max_status_bytes}"
+            )
+        return json.loads(encoded)
+
+    def update(self, name: str, status: Mapping[str, Any]) -> int:
+        """Replace one system status and return the registry revision."""
+
+        normalized = self._validate_name(name)
+        cloned = self._clone_status(status)
+        now = float(self._clock())
+        with self._lock:
+            if (
+                normalized not in self._systems
+                and len(self._systems) >= self._max_systems
+            ):
+                raise ValueError(f"registry is limited to {self._max_systems} systems")
+            self._revision += 1
+            self._systems[normalized] = {
+                "revision": self._revision,
+                "updated_at_s": now,
+                "status": cloned,
+            }
+            self._updated_at_s = now
+            return self._revision
+
+    def remove(self, name: str) -> bool:
+        """Remove one system status if present."""
+
+        normalized = self._validate_name(name)
+        with self._lock:
+            if normalized not in self._systems:
+                return False
+            del self._systems[normalized]
+            self._revision += 1
+            self._updated_at_s = float(self._clock())
+            return True
+
+    def snapshot(self) -> dict[str, Any]:
+        """Return a detached JSON-compatible view of all published statuses."""
+
+        with self._lock:
+            revision = self._revision
+            updated_at_s = self._updated_at_s
+            systems = json.loads(json.dumps(self._systems, separators=(",", ":")))
+        return {
+            "ts": float(self._clock()),
+            "revision": revision,
+            "updated_at_s": updated_at_s,
+            "system_count": len(systems),
+            "systems": systems,
+        }
+
+
+def runtime_systems_snapshot(state: Any) -> dict[str, Any]:
+    """Read a registry from a server state without requiring a concrete type."""
+
+    registry = getattr(state, "runtime_systems", None)
+    snapshot = getattr(registry, "snapshot", None)
+    if callable(snapshot):
+        payload = snapshot()
+        if isinstance(payload, Mapping):
+            return dict(payload)
+    now = time.time()
+    return {
+        "ts": now,
+        "revision": 0,
+        "updated_at_s": now,
+        "system_count": 0,
+        "systems": {},
+    }
+
+
+def install_runtime_systems_endpoint(
+    app: Any, state: Any, *, refresh: Callable[[], None] | None = None
+) -> None:
+    """Install the read-only runtime systems endpoint on a FastAPI app."""
+
+    @app.get("/v1/mtplx/systems")
+    def mtplx_runtime_systems() -> dict[str, Any]:
+        if refresh is not None:
+            refresh()
+        return runtime_systems_snapshot(state)
+
+
+__all__ = [
+    "RuntimeSystemsRegistry",
+    "install_runtime_systems_endpoint",
+    "runtime_systems_snapshot",
+]

mtplx/server/openai.py

@@ -115,6 +115,10 @@
     resolve_gemma4_pair_paths,
 )
 from mtplx.model_scheduler import ModelWorkScheduler
+from mtplx.runtime_systems import (
+    RuntimeSystemsRegistry,
+    install_runtime_systems_endpoint,
+)
 from mtplx.server.hyper import HYPER_ADMISSION_CAP, HyperAdmissionGate
 from mtplx.reasoning_effort import (
     REASONING_EFFORT_CHOICES,
@@ -2869,6 +2873,7 @@ def __init__(self, args: argparse.Namespace) -> None:
         _validate_mtp_batch_settings(args)
         _validate_hyper_settings(args)
         self.args = args
+        self.runtime_systems = RuntimeSystemsRegistry()
         try:
             args.paged_kv_quantization = normalize_paged_kv_quantization(
                 getattr(args, "paged_kv_quantization", "off")
@@ -28706,6 +28711,12 @@ async def lifespan(_app: FastAPI):
 
     app = FastAPI(title="MTPLX OpenAI-compatible server", lifespan=lifespan)
     app.state.mtplx = state
+    from mtplx.server.runtime_status import ServingStatusProvider
+
+    if getattr(state, "runtime_systems", None) is None:
+        state.runtime_systems = RuntimeSystemsRegistry()
+    serving_status = ServingStatusProvider(state)
+    install_runtime_systems_endpoint(app, state, refresh=serving_status.publish)
     # Registered before the auth middleware below so auth stays outermost
     # (the most recently added Starlette middleware runs first): fans only
     # ramp for requests that passed the API-key and rate-limit gates.

mtplx/server/runtime_status.py

@@ -0,0 +1,82 @@
+"""Read-only serving-status provider for the generic runtime systems registry.
+
+Only bounded aggregate scalars cross this boundary. No model lock, allocator,
+cache mutation, configuration changes, prompt content, paths, or client IDs.
+"""
+
+from __future__ import annotations
+
+import threading
+from typing import Any
+
+
+def _counter(value: Any) -> int | None:
+    # Do not stringify or invoke arbitrary object conversion for telemetry.
+    return value if type(value) is int and 0 <= value <= 2**63 - 1 else None
+
+
+class ServingStatusProvider:
+    """Refresh from the existing server's CPU-side state, not a second owner."""
+
+    def __init__(self, state: Any) -> None:
+        self._state = state
+        self._lock = threading.Lock()
+
+    def publish(self) -> None:
+        # Serialize collection/publication so an older read cannot replace a newer one.
+        with self._lock:
+            try:
+                status = self._read()
+            except (AttributeError, RuntimeError, TypeError, ValueError, OSError):
+                # Never leave a prior healthy snapshot visible after a failed refresh.
+                # Error text may contain paths or credentials, so only a fixed code escapes.
+                status = {
+                    "available": False,
+                    "enabled": False,
+                    "wired": True,
+                    "phase": "unavailable",
+                    "reason": "status_read_failed",
+                }
+            self._state.runtime_systems.update("serving", status)
+
+    def _read(self) -> dict[str, Any]:
+        state = self._state
+        runtime = getattr(state, "runtime", None)
+        released = getattr(state, "aime_parent_runtime_released", False) is True
+        available = runtime is not None and not released
+        foreground = getattr(state, "foreground_count", None)
+        active = _counter(
+            foreground()
+            if callable(foreground)
+            else getattr(state, "foreground_active", None)
+        )
+        in_flight = getattr(getattr(state, "dashboard", None), "in_flight", None)
+        count = getattr(in_flight, "count", None)
+        if callable(count):
+            dashboard_active = _counter(count())
+            if dashboard_active is not None:
+                active = max(active or 0, dashboard_active)
+        mode = getattr(getattr(state, "args", None), "generation_mode", None)
+        mode = mode if type(mode) is str and mode in {"ar", "mtp"} else None
+        return {
+            "available": available,
+            "enabled": available,
+            "wired": True,
+            "phase": (
+                "unavailable"
+                if not available
+                else "unknown"
+                if active is None
+                else "busy"
+                if active
+                else "idle"
+            ),
+            "generation_mode": mode,
+            "active_requests": active,
+            "requests_completed": _counter(getattr(state, "requests_completed", None)),
+            "requests_cancelled": _counter(getattr(state, "requests_cancelled", None)),
+            "mtp_enabled": getattr(runtime, "mtp_enabled", False) is True
+            if available
+            else False,
+            "sample_scope": "aggregate_scalars_not_a_transactional_snapshot",
+        }