diff --git a/docs/runtime-systems.md b/docs/runtime-systems.md
new file mode 100644
index 000000000..fda95f889
--- /dev/null
+++ b/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.
diff --git a/mtplx/runtime_systems.py b/mtplx/runtime_systems.py
new file mode 100644
index 000000000..8b2261171
--- /dev/null
+++ b/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",
+]
diff --git a/mtplx/server/openai.py b/mtplx/server/openai.py
index 2fbcdd0e6..6221c30ac 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,
@@ -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.
diff --git a/mtplx/server/runtime_status.py b/mtplx/server/runtime_status.py
new file mode 100644
index 000000000..7224236b7
--- /dev/null
+++ b/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",
+        }
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_serving_status.py b/tests/test_runtime_serving_status.py
new file mode 100644
index 000000000..4fcffeeee
--- /dev/null
+++ b/tests/test_runtime_serving_status.py
@@ -0,0 +1,173 @@
+"""Live provider contract, including the actual production FastAPI wiring."""
+
+import json
+import subprocess
+import sys
+from concurrent.futures import ThreadPoolExecutor
+from pathlib import Path
+from types import SimpleNamespace
+
+from fastapi.testclient import TestClient
+
+from mtplx.runtime_systems import RuntimeSystemsRegistry
+from mtplx.server.runtime_status import ServingStatusProvider
+
+
+def state():
+    return SimpleNamespace(
+        runtime_systems=RuntimeSystemsRegistry(),
+        runtime=SimpleNamespace(mtp_enabled=True),
+        args=SimpleNamespace(
+            generation_mode="mtp", model="/private/model", api_key="secret"
+        ),
+        foreground_active=0,
+        requests_completed=0,
+        requests_cancelled=0,
+    )
+
+
+def status(value):
+    return value.runtime_systems.snapshot()["systems"]["serving"]["status"]
+
+
+def test_provider_tracks_existing_serving_lifecycle_without_mutating_it():
+    value = state()
+    provider = ServingStatusProvider(value)
+    provider.publish()
+    assert status(value)["phase"] == "idle"
+    value.foreground_active = 2
+    value.requests_completed = 7
+    value.requests_cancelled = 1
+    provider.publish()
+    assert status(value)["phase"] == "busy"
+    assert status(value)["active_requests"] == 2
+    assert status(value)["requests_completed"] == 7
+    assert status(value)["requests_cancelled"] == 1
+    assert value.foreground_active == 2
+    value.runtime = None
+    provider.publish()
+    assert status(value)["available"] is False
+    assert status(value)["mtp_enabled"] is False
+
+
+def test_provider_counts_dashboard_and_foreground_without_model_lock():
+    value = state()
+    value.foreground_count = lambda: 1
+    value.dashboard = SimpleNamespace(in_flight=SimpleNamespace(count=lambda: 3))
+
+    class ForbiddenLock:
+        def __enter__(self):
+            raise AssertionError("Provider must never acquire the model lock")
+
+    value.model_lock = ForbiddenLock()
+    ServingStatusProvider(value).publish()
+    assert status(value)["active_requests"] == 3
+
+
+def test_failed_refresh_replaces_stale_status_and_redacts_errors():
+    value = state()
+    provider = ServingStatusProvider(value)
+    provider.publish()
+
+    def broken():
+        raise RuntimeError("/private/path?api_key=secret")
+
+    value.foreground_count = broken
+    provider.publish()
+    assert status(value) == {
+        "available": False,
+        "enabled": False,
+        "wired": True,
+        "phase": "unavailable",
+        "reason": "status_read_failed",
+    }
+    assert "secret" not in json.dumps(status(value))
+    value.foreground_count = lambda: 0
+    provider.publish()
+    assert status(value)["phase"] == "idle"
+
+
+def test_unknown_values_are_not_fabricated_as_valid_metrics():
+    value = state()
+    value.foreground_active = None
+    value.requests_completed = True
+    value.requests_cancelled = -1
+    value.args.generation_mode = "secret"
+    ServingStatusProvider(value).publish()
+    result = status(value)
+    assert result["phase"] == "unknown"
+    assert result["active_requests"] is None
+    assert result["requests_completed"] is None
+    assert result["requests_cancelled"] is None
+    assert result["generation_mode"] is None
+    assert "private" not in json.dumps(result)
+    assert "secret" not in json.dumps(result)
+
+
+def test_released_runtime_is_not_advertised_available():
+    value = state()
+    value.aime_parent_runtime_released = True
+    ServingStatusProvider(value).publish()
+    assert status(value)["phase"] == "unavailable"
+    assert status(value)["enabled"] is False
+
+
+def test_concurrent_refresh_is_bounded_and_snapshots_are_detached():
+    value = state()
+    provider = ServingStatusProvider(value)
+    with ThreadPoolExecutor(max_workers=8) as pool:
+        list(pool.map(lambda _: provider.publish(), range(64)))
+    snapshot = value.runtime_systems.snapshot()
+    assert snapshot["system_count"] == 1
+    assert snapshot["revision"] == 64
+    snapshot["systems"]["serving"]["status"]["active_requests"] = 999
+    assert status(value)["active_requests"] == 0
+
+
+def test_production_app_refreshes_real_state_and_keeps_auth_boundary():
+    from test_server_openai import _fake_state
+
+    from mtplx.server.openai import create_app
+
+    value = _fake_state(api_key="test-key")
+    value.foreground_count = lambda: value.foreground_active
+    value.foreground_active = 0
+    value.requests_completed = 4
+    client = TestClient(create_app(value))
+    assert client.get("/v1/mtplx/systems").status_code == 401
+    assert value.runtime_systems.snapshot()["system_count"] == 0
+    response = client.get(
+        "/v1/mtplx/systems", headers={"Authorization": "Bearer test-key"}
+    )
+    assert response.status_code == 200
+    assert response.json()["systems"]["serving"]["status"]["requests_completed"] == 4
+    value.foreground_active = 3
+    value.requests_completed = 5
+    response = client.get(
+        "/v1/mtplx/systems", headers={"Authorization": "Bearer test-key"}
+    )
+    payload = response.json()["systems"]["serving"]["status"]
+    assert payload["phase"] == "busy"
+    assert payload["active_requests"] >= 3
+    assert payload["requests_completed"] == 5
+    assert "test-key" not in response.text
+
+
+def test_provider_import_does_not_import_mlx():
+    root = Path(__file__).resolve().parents[1]
+    result = subprocess.run(
+        [
+            sys.executable,
+            "-c",
+            (
+                "import sys; from mtplx.server.runtime_status import ServingStatusProvider; "
+                "assert not any(n == 'mlx' or n.startswith('mlx.') for n in sys.modules)"
+            ),
+        ],
+        cwd=root,
+        capture_output=True,
+        text=True,
+        timeout=15,
+        check=False,
+    )
+    assert result.returncode == 0, result.stderr
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}