From 08e16c8bc80b5e85cafcd18cb1e7d4008c57b043 Mon Sep 17 00:00:00 2001
From: Philip John Basile <PBasile@Basilecom.com>
Date: Wed, 26 Aug 2026 09:20:53 -0400
Subject: [PATCH] feat(server): add runtime systems status registry

---
 mtplx/runtime_systems.py         | 150 +++++++++++++++++++++++++++++++
 mtplx/server/openai.py           |   6 ++
 scripts/bench_runtime_systems.py | 124 +++++++++++++++++++++++++
 tests/test_runtime_systems.py    | 105 ++++++++++++++++++++++
 4 files changed, 385 insertions(+)
 create mode 100644 mtplx/runtime_systems.py
 create mode 100644 scripts/bench_runtime_systems.py
 create mode 100644 tests/test_runtime_systems.py

diff --git a/mtplx/runtime_systems.py b/mtplx/runtime_systems.py
new file mode 100644
index 000000000..c3ebc1436
--- /dev/null
+++ b/mtplx/runtime_systems.py
@@ -0,0 +1,150 @@
+"""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) -> None:
+    """Install the read-only runtime systems endpoint on a FastAPI app."""
+
+    @app.get("/v1/mtplx/systems")
+    def mtplx_runtime_systems() -> dict[str, Any]:
+        return runtime_systems_snapshot(state)
+
+
+__all__ = [
+    "RuntimeSystemsRegistry",
+    "install_runtime_systems_endpoint",
+    "runtime_systems_snapshot",
+]
diff --git a/mtplx/server/openai.py b/mtplx/server/openai.py
index 73590798c..dede5ec49 100644
--- a/mtplx/server/openai.py
+++ b/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,
@@ -2236,6 +2240,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")
@@ -26888,6 +26893,7 @@ async def lifespan(_app: FastAPI):
 
     app = FastAPI(title="MTPLX OpenAI-compatible server", lifespan=lifespan)
     app.state.mtplx = state
+    install_runtime_systems_endpoint(app, state)
     # 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.
diff --git a/scripts/bench_runtime_systems.py b/scripts/bench_runtime_systems.py
new file mode 100644
index 000000000..2eb4e8c95
--- /dev/null
+++ b/scripts/bench_runtime_systems.py
@@ -0,0 +1,124 @@
+#!/usr/bin/env python3
+"""Measure runtime status update, snapshot, and HTTP endpoint overhead."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import platform
+import statistics
+import time
+from pathlib import Path
+from types import SimpleNamespace
+from typing import Any
+
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from mtplx.runtime_systems import (
+    RuntimeSystemsRegistry,
+    install_runtime_systems_endpoint,
+)
+
+
+def percentile(samples_ns: list[int], percentile_value: float) -> float:
+    ordered = sorted(samples_ns)
+    index = min(len(ordered) - 1, int((len(ordered) - 1) * percentile_value))
+    return ordered[index] / 1_000.0
+
+
+def summarize(samples_ns: list[int]) -> dict[str, float]:
+    elapsed_s = sum(samples_ns) / 1_000_000_000.0
+    return {
+        "p50_us": percentile(samples_ns, 0.50),
+        "p95_us": percentile(samples_ns, 0.95),
+        "p99_us": percentile(samples_ns, 0.99),
+        "mean_us": statistics.fmean(samples_ns) / 1_000.0,
+        "operations_per_s": len(samples_ns) / elapsed_s,
+    }
+
+
+def measure_call(callback: Any, iterations: int) -> list[int]:
+    samples: list[int] = []
+    for index in range(iterations):
+        started = time.perf_counter_ns()
+        callback(index)
+        samples.append(time.perf_counter_ns() - started)
+    return samples
+
+
+def run(iterations: int, endpoint_requests: int, system_count: int) -> dict[str, Any]:
+    registry = RuntimeSystemsRegistry(max_systems=max(128, system_count))
+    for index in range(system_count):
+        registry.update(
+            f"system.{index}",
+            {
+                "available": True,
+                "enabled": index % 2 == 0,
+                "metrics": {"requests": index, "queue_depth": index % 4},
+            },
+        )
+
+    update_samples = measure_call(
+        lambda index: registry.update(
+            "system.0",
+            {
+                "available": True,
+                "enabled": True,
+                "metrics": {"requests": index, "queue_depth": index % 4},
+            },
+        ),
+        iterations,
+    )
+    snapshot_samples = measure_call(lambda _index: registry.snapshot(), iterations)
+
+    app = FastAPI()
+    install_runtime_systems_endpoint(app, SimpleNamespace(runtime_systems=registry))
+    with TestClient(app) as client:
+        endpoint_samples = measure_call(
+            lambda _index: client.get("/v1/mtplx/systems").raise_for_status(),
+            endpoint_requests,
+        )
+        response = client.get("/v1/mtplx/systems")
+
+    payload = response.json()
+    return {
+        "environment": {
+            "python": platform.python_version(),
+            "platform": platform.platform(),
+        },
+        "parameters": {
+            "iterations": iterations,
+            "endpoint_requests": endpoint_requests,
+            "system_count": system_count,
+        },
+        "update": summarize(update_samples),
+        "snapshot": summarize(snapshot_samples),
+        "http_get": summarize(endpoint_samples),
+        "correctness": {
+            "http_status": response.status_code,
+            "reported_system_count": payload["system_count"],
+            "reported_revision": payload["revision"],
+        },
+    }
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--iterations", type=int, default=10_000)
+    parser.add_argument("--endpoint-requests", type=int, default=1_000)
+    parser.add_argument("--system-count", type=int, default=16)
+    parser.add_argument("--output", type=Path)
+    args = parser.parse_args()
+    if args.iterations < 1 or args.endpoint_requests < 1 or args.system_count < 1:
+        parser.error("all counts must be positive")
+
+    result = run(args.iterations, args.endpoint_requests, args.system_count)
+    encoded = json.dumps(result, indent=2, sort_keys=True)
+    if args.output is not None:
+        args.output.write_text(encoded + "\n", encoding="utf-8")
+    print(encoded)
+
+
+if __name__ == "__main__":
+    main()
diff --git a/tests/test_runtime_systems.py b/tests/test_runtime_systems.py
new file mode 100644
index 000000000..e3b3f37b1
--- /dev/null
+++ b/tests/test_runtime_systems.py
@@ -0,0 +1,105 @@
+from __future__ import annotations
+
+from concurrent.futures import ThreadPoolExecutor
+from types import SimpleNamespace
+
+import pytest
+
+from mtplx.runtime_systems import (
+    RuntimeSystemsRegistry,
+    install_runtime_systems_endpoint,
+    runtime_systems_snapshot,
+)
+
+
+def test_update_replaces_one_provider_neutral_status() -> None:
+    registry = RuntimeSystemsRegistry(clock=lambda: 12.5)
+    source = {"enabled": True, "metrics": {"hits": 3}}
+
+    revision = registry.update("cache.primary", source)
+    source["metrics"]["hits"] = 99
+
+    payload = registry.snapshot()
+    assert revision == 1
+    assert payload == {
+        "ts": 12.5,
+        "revision": 1,
+        "updated_at_s": 12.5,
+        "system_count": 1,
+        "systems": {
+            "cache.primary": {
+                "revision": 1,
+                "updated_at_s": 12.5,
+                "status": {"enabled": True, "metrics": {"hits": 3}},
+            }
+        },
+    }
+
+
+def test_snapshot_is_detached_from_registry_state() -> None:
+    registry = RuntimeSystemsRegistry()
+    registry.update("scheduler", {"queue": {"depth": 2}})
+
+    first = registry.snapshot()
+    first["systems"]["scheduler"]["status"]["queue"]["depth"] = 500
+
+    assert registry.snapshot()["systems"]["scheduler"]["status"]["queue"]["depth"] == 2
+
+
+def test_registry_bounds_names_payloads_and_cardinality() -> None:
+    registry = RuntimeSystemsRegistry(max_systems=1, max_status_bytes=16)
+    registry.update("valid-name", {"ok": True})
+
+    with pytest.raises(ValueError, match="limited"):
+        registry.update("another", {})
+    with pytest.raises(ValueError, match="system name"):
+        registry.update("not valid", {})
+    with pytest.raises(ValueError, match="max_status_bytes"):
+        registry.update("valid-name", {"value": "x" * 20})
+    with pytest.raises(ValueError, match="JSON-compatible"):
+        registry.update("valid-name", {"value": object()})
+
+
+def test_concurrent_updates_produce_a_consistent_snapshot() -> None:
+    registry = RuntimeSystemsRegistry(max_systems=16)
+
+    def publish(index: int) -> None:
+        for value in range(100):
+            registry.update(f"worker.{index}", {"value": value})
+
+    with ThreadPoolExecutor(max_workers=8) as pool:
+        list(pool.map(publish, range(8)))
+
+    payload = registry.snapshot()
+    assert payload["revision"] == 800
+    assert payload["system_count"] == 8
+    assert all(item["status"]["value"] == 99 for item in payload["systems"].values())
+
+
+def test_remove_advances_revision_only_when_present() -> None:
+    registry = RuntimeSystemsRegistry()
+    registry.update("worker", {"ready": True})
+
+    assert registry.remove("worker") is True
+    assert registry.remove("worker") is False
+    assert registry.snapshot()["revision"] == 2
+
+
+def test_missing_registry_returns_an_empty_snapshot() -> None:
+    payload = runtime_systems_snapshot(SimpleNamespace())
+    assert payload["system_count"] == 0
+    assert payload["systems"] == {}
+
+
+def test_http_endpoint_returns_the_live_registry_snapshot() -> None:
+    fastapi = pytest.importorskip("fastapi")
+    testclient = pytest.importorskip("fastapi.testclient")
+    state = SimpleNamespace(runtime_systems=RuntimeSystemsRegistry())
+    state.runtime_systems.update("decoder", {"ready": True})
+    app = fastapi.FastAPI()
+    install_runtime_systems_endpoint(app, state)
+
+    response = testclient.TestClient(app).get("/v1/mtplx/systems")
+
+    assert response.status_code == 200
+    assert response.json()["systems"]["decoder"]["status"] == {"ready": True}