feat(ops): expose worker telemetry and Mac MLX equivalence preset

Co-authored-by: FluffyAIcode <FluffyAIcode@users.noreply.github.com>

Author: cursoragent

inference_engine/bridge/manifest.py

@@ -104,6 +104,22 @@ def _harness_preset(
 PRESETS: Dict[str, Preset] = {
     p.name: p
     for p in (
+        Preset(
+            name="mlx-prefill-snapshot-equivalence",
+            description="ADR 0017 real-model gate: local MLX prefill vs "
+                        "snapshot export/import must preserve continuation "
+                        "logits, argmax, retained KV tokens and one decode step.",
+            command_templates=(
+                (
+                    "python3", "-m", "pytest",
+                    "-m", "integration",
+                    "tests/integration/test_prefill_snapshot_mlx_equivalence.py",
+                    "-q",
+                ),
+            ),
+            timeout_minutes=45,
+            validate_reports=False,
+        ),
         Preset(
             name="mlx-distributed-dflash-e2e-inproc",
             description="Real-model distributed DFlash+f_θ E2E (in-process): loads "

inference_engine/network/state.py

@@ -95,6 +95,7 @@ def nodes(self) -> list[dict[str, Any]]:
         for card in self.registry.snapshot():
             registration = registrations.get(card.node_id, {})
             cache = card.caches[0] if card.caches else None
+            worker = card.prefill_workers[0] if card.prefill_workers else None
             endpoint = sorted(
                 card.endpoints,
                 key=lambda item: item.priority,
@@ -132,6 +133,19 @@ def nodes(self) -> list[dict[str, Any]]:
                     }
                     if cache else None
                 ),
+                "prefill_worker": (
+                    {
+                        "address": worker.worker_address,
+                        "max_concurrent_jobs": worker.max_concurrent_jobs,
+                        "inflight_jobs": worker.inflight_jobs,
+                        "queued_jobs": worker.queued_jobs,
+                        "queued_tokens": worker.queued_tokens,
+                        "load": worker.load,
+                        "tokens_per_second": worker.tokens_per_second_prefill,
+                        "ram_bytes_free": worker.ram_bytes_free,
+                    }
+                    if worker else None
+                ),
                 "endpoint": (
                     {
                         "address": endpoint[0].address,
@@ -160,6 +174,7 @@ def nodes(self) -> list[dict[str, Any]]:
                     "memory_bytes": 0,
                     "models": [],
                     "cache": None,
+                    "prefill_worker": None,
                     "endpoint": {
                         "address": registration["address"],
                         "network": "pending",

tests/inference_engine/bridge/test_manifest.py

@@ -96,6 +96,7 @@ def test_allowlist_contains_exactly_the_documented_presets():
         "mlx-kakeya-launcher-full",
         "mlx-kakeya-launcher-smoke",
         "mlx-multitenant-pressure",
+        "mlx-prefill-snapshot-equivalence",
         "mlx-upgrade",
         "mlx-upstream-batch-probe",
         "pytest-path",

tests/inference_engine/network/test_network_state.py

@@ -6,6 +6,7 @@
     CapabilityRegistry,
     NodeCapability,
     NodeEndpoint,
+    PrefillWorkerCapability,
 )
 from inference_engine.distributed.prefill_cache import PrefixCacheStore
 from inference_engine.network.state import NetworkState
@@ -27,6 +28,12 @@ def _state(tmp_path):
             ),
         ),
         endpoints=(NodeEndpoint("head:2", "thunderbolt", 100, 0.4),),
+        prefill_workers=(PrefillWorkerCapability(
+            compatibility,
+            worker_address="head:3",
+            queued_tokens=128,
+            tokens_per_second_prefill=32,
+        ),),
     )
     return NetworkState(
         CapabilityRegistry(self_card=card),