feat(system): add Worker 006 telemetry and network profile

experiments/perception/worker/run_e15_shadow_inference.py

@@ -1138,7 +1138,11 @@ def _load_models(args: argparse.Namespace, common: dict[str, Any]) -> _LoadedMod
     )
 
 
-def run(args: argparse.Namespace, loaded: _LoadedModels | None = None) -> int:
+def run(
+    args: argparse.Namespace,
+    loaded: _LoadedModels | None = None,
+    runtime_state: dict[str, Any] | None = None,
+) -> int:
     import av
     import torch
     import transformers
@@ -1149,6 +1153,18 @@ def run(args: argparse.Namespace, loaded: _LoadedModels | None = None) -> int:
     token = sys.stdin.readline().strip() if args.token_stdin else args.token
     if not token or len(token) < 40:
         raise RuntimeError("LAB E15 shadow token is missing")
+
+    def report_stage(stage: str, frame_index: int | None = None) -> None:
+        if runtime_state is None:
+            return
+        runtime_state["current_stage"] = stage
+        runtime_state["stage_observed_at_utc"] = (
+            datetime.now(UTC).isoformat().replace("+00:00", "Z")
+        )
+        if frame_index is not None:
+            runtime_state["active_frame_index"] = frame_index
+
+    report_stage("preprocessing")
     common = _common(args) if loaded is None else loaded.common
     live = common["live"]
     e14 = common["e14"]
@@ -1215,6 +1231,7 @@ def run(args: argparse.Namespace, loaded: _LoadedModels | None = None) -> int:
     def on_decoded(frame: DecodedCameraFrame) -> None:
         nonlocal decoded_frame_count
         try:
+            report_stage("camera-decode", frame.frame_index)
             if not first_camera_epoch_ns:
                 first_camera_epoch_ns.append(frame.metadata.captured_at_epoch_ns)
             last_camera_epoch_ns[:] = [frame.metadata.captured_at_epoch_ns]
@@ -1383,6 +1400,7 @@ def run(args: argparse.Namespace, loaded: _LoadedModels | None = None) -> int:
         )
         semantic_thread.start()
         decoder.start()
+        report_stage("source-ingress")
         receiver_thread = threading.Thread(
             target=_receiver,
             kwargs={
@@ -1429,10 +1447,13 @@ def run(args: argparse.Namespace, loaded: _LoadedModels | None = None) -> int:
             latency["decode_age_ms"].append(float(envelope.decode_ms))
             latency["queue_wait_ms"].append(max(0.0, (started - envelope.decoded_monotonic) * 1000))
             try:
+                report_stage("preprocessing", envelope.frame_index)
                 detector_started = time.perf_counter()
                 tensor = _preprocess(envelope.image, valid_mask, detector)
+                report_stage("detector", envelope.frame_index)
                 output_tensor, _request_ms = _infer(args.triton_url, detector["model"], tensor)
                 detections, _rejected = _detections(output_tensor, detector, valid_mask)
+                report_stage("tracking", envelope.frame_index)
                 tracks = tracker.update(detections, envelope.frame_index)
                 latency["detector_ms"].append((time.perf_counter() - detector_started) * 1000)
 
@@ -1463,6 +1484,7 @@ def run(args: argparse.Namespace, loaded: _LoadedModels | None = None) -> int:
                     ).reshape((-1, 3))
                     position = binding.pose.position_xyz
                     quaternion = binding.pose.orientation_xyzw
+                    report_stage("sensor-fusion", envelope.frame_index)
                     projection_started = time.perf_counter()
                     pixels, depths, source_indices, points_lidar = project_points(
                         points_map,
@@ -1533,6 +1555,7 @@ def run(args: argparse.Namespace, loaded: _LoadedModels | None = None) -> int:
                 if temporal_stabilizer is None:
                     fusion_objects = raw_fusion_objects
                 else:
+                    report_stage("temporal-state", envelope.frame_index)
                     temporal_started = time.perf_counter()
                     fusion_objects = temporal_stabilizer.update(
                         frame_index=envelope.frame_index,
@@ -1619,6 +1642,7 @@ def run(args: argparse.Namespace, loaded: _LoadedModels | None = None) -> int:
                     objects=fusion_objects,
                     delivery=world["delivery"],
                 )
+                report_stage("result-publication", envelope.frame_index)
                 try:
                     result_queue.put_nowait(live_result)
                 except queue.Full:
@@ -1627,6 +1651,7 @@ def run(args: argparse.Namespace, loaded: _LoadedModels | None = None) -> int:
                         result_queue.task_done()
                         transport.results_dropped += 1
                     result_queue.put_nowait(live_result)
+                report_stage("source-ingress", envelope.frame_index)
             except Exception:
                 detector_failures += 1
                 raise
@@ -2063,7 +2088,15 @@ def serve(args: argparse.Namespace) -> int:
     loaded = _load_models(args, common)
     model_load_seconds = time.perf_counter() - load_started
     run_lock = threading.Lock()
-    state = {"busy": False, "completed_runs": 0, "failed_runs": 0}
+    state: dict[str, Any] = {
+        "busy": False,
+        "completed_runs": 0,
+        "failed_runs": 0,
+        "active_request_id": None,
+        "active_frame_index": None,
+        "current_stage": None,
+        "stage_observed_at_utc": None,
+    }
 
     class Handler(BaseHTTPRequestHandler):
         server_version = "MissionCorePersistentPerception/1"
@@ -2100,6 +2133,10 @@ def serve(args: argparse.Namespace) -> int:
                     "model_load_seconds": model_load_seconds,
                     "completed_runs": state["completed_runs"],
                     "failed_runs": state["failed_runs"],
+                    "active_request_id": state["active_request_id"],
+                    "active_frame_index": state["active_frame_index"],
+                    "current_stage": state["current_stage"],
+                    "stage_observed_at_utc": state["stage_observed_at_utc"],
                     "authority": common["live"]["authority"],
                     "gpu": torch.cuda.get_device_name(),
                 },
@@ -2129,7 +2166,8 @@ def serve(args: argparse.Namespace) -> int:
                 request_id = str(document.get("request_id", "invalid"))
                 run_args = _persistent_run_arguments(args, document)
                 document["token"] = None
-                exit_code = run(run_args, loaded)
+                state["active_request_id"] = request_id
+                exit_code = run(run_args, loaded, state)
                 state["completed_runs"] += 1
                 self._send(
                     200,
@@ -2164,6 +2202,10 @@ def serve(args: argparse.Namespace) -> int:
                 )
             finally:
                 state["busy"] = False
+                state["active_request_id"] = None
+                state["active_frame_index"] = None
+                state["current_stage"] = None
+                state["stage_observed_at_utc"] = None
                 run_lock.release()
 
     server = ThreadingHTTPServer((args.listen_host, args.listen_port), Handler)